revert: remove file-based configuration and multi-spec support
CI / Build and push docker image (push) Successful in 52s

This reverts commit 9185b5fa05 while
preserving the build work done afterwards: setuptools-scm versioning with
uv cache keys, the git-enabled Docker builder (full checkout, credentials
excluded), and full-history/tag checkout in CI.

Configuration returns to environment variables with a single OpenAPI
document; .env.example and the env-based Docker/Compose setup are
restored.
This commit is contained in:
2026-10-22 04:54:47 +00:00
parent 0063e7c309
commit 827549c0ec
10 changed files with 326 additions and 743 deletions
+1 -2
View File
@@ -6,8 +6,7 @@ __pycache__/
.mypy_cache/ .mypy_cache/
.env .env
.env.* .env.*
xmcp.yml !.env.example
xmcp.local.yml
# Generated at build time # Generated at build time
src/xmcp/_version.py src/xmcp/_version.py
+46
View File
@@ -0,0 +1,46 @@
# Copy to .env and adjust. All values are read at container/process start.
# --- OpenAPI document (required) ---------------------------------------------
# http(s) URL of a JSON/YAML OpenAPI document, or a local file path
# (e.g. /spec/openapi.json when the document is mounted into the container).
OPENAPI_SPEC_URL=https://petstore3.swagger.io/api/v3/openapi.json
# Bearer token used only when fetching a remote spec (leave unset for public
# documents).
#OPENAPI_SPEC_TOKEN=
# --- Target API ---------------------------------------------------------------
# Override the base URL taken from the spec's servers[0].url. Required when the
# spec only has a relative server URL and the spec is a local file, or when the
# API lives on a different host than the spec.
#API_BASE_URL=https://api.example.com
# Credential sent to the target API on every tool call.
#API_TOKEN=change-me
#API_AUTH_HEADER=Authorization
# Send the raw token with no scheme prefix by setting this to an empty value:
#API_AUTH_SCHEME=
# HTTP header treated as the per-request API key of the MCP client. When
# present it takes precedence over API_TOKEN. Set to an empty value to disable
# the mapping (default: X-API-Key).
#MCP_API_KEY_HEADER=X-API-Key
# --- MCP server ---------------------------------------------------------------
# stdio or http (streamable-http is accepted as an alias); default: stdio
MCP_TRANSPORT=http
MCP_HOST=0.0.0.0
MCP_PORT=8000
#MCP_SERVER_NAME=My API
# Comma-separated allowlists of tool names and OpenAPI tags to expose. Unset or
# empty disables that filter; unknown entries are ignored with a startup
# warning. When both are set, a tool must match both.
#MCP_TOOLS_INCLUDED=getPetById,findPetsByStatus
#MCP_TOOLS_TAGS_INCLUDED=pet
# Timeout in seconds for fetching the spec and calling the API (default: 30).
#HTTP_TIMEOUT=30
# DEBUG, INFO, WARNING, ERROR, CRITICAL ... (default: INFO)
#LOG_LEVEL=INFO
+2 -3
View File
@@ -5,11 +5,10 @@ __pycache__/
# uv / virtual environments # uv / virtual environments
.venv/ .venv/
# Local configuration (may contain secrets) # Local configuration
xmcp.yml
xmcp.local.yml
.env .env
.env.* .env.*
!.env.example
# Generated version metadata and build artifacts # Generated version metadata and build artifacts
src/xmcp/_version.py src/xmcp/_version.py
+7 -5
View File
@@ -32,11 +32,13 @@ FROM python:3.14-alpine3.24 AS runtime
ENV PYTHONDONTWRITEBYTECODE=1 \ ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \ PYTHONUNBUFFERED=1 \
PATH="/app/.venv/bin:$PATH" PATH="/app/.venv/bin:$PATH" \
MCP_TRANSPORT=http \
MCP_HOST=0.0.0.0 \
MCP_PORT=8000
RUN addgroup -S -g 1001 xmcp \ RUN addgroup -S -g 1001 xmcp \
&& adduser -S -D -H -u 1001 -G xmcp xmcp \ && adduser -S -D -H -u 1001 -G xmcp xmcp
&& mkdir -p /etc/xmcp
WORKDIR /app WORKDIR /app
COPY --from=builder --chown=xmcp:xmcp /app/.venv /app/.venv COPY --from=builder --chown=xmcp:xmcp /app/.venv /app/.venv
@@ -45,8 +47,8 @@ USER xmcp
EXPOSE 8000 EXPOSE 8000
# Probes the MCP HTTP port when the mounted configuration uses transport http. # In HTTP mode, probe the MCP port; stdio containers are always healthy.
HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \ HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
CMD ["python", "-m", "xmcp.healthcheck"] CMD ["python", "-c", "import os, socket, sys\nif os.environ.get('MCP_TRANSPORT', 'stdio') != 'http':\n sys.exit(0)\ns = socket.socket()\ns.settimeout(3)\nsys.exit(s.connect_ex(('127.0.0.1', int(os.environ.get('MCP_PORT', '8000')))))"]
CMD ["python", "-m", "xmcp"] CMD ["python", "-m", "xmcp"]
+118 -147
View File
@@ -10,148 +10,104 @@
[![Docker](https://img.shields.io/badge/Docker-image-2496ED?logo=docker&logoColor=white&style=flat-square)](https://www.docker.com/) [![Docker](https://img.shields.io/badge/Docker-image-2496ED?logo=docker&logoColor=white&style=flat-square)](https://www.docker.com/)
[![uv](https://img.shields.io/badge/uv-managed-DE5FE9?style=flat-square)](https://docs.astral.sh/uv/) [![uv](https://img.shields.io/badge/uv-managed-DE5FE9?style=flat-square)](https://docs.astral.sh/uv/)
Turns one or more [OpenAPI](https://spec.openapis.org/oas/latest.html) documents Turns any [OpenAPI](https://spec.openapis.org/oas/latest.html) document
(JSON or YAML) into a ready-to-run [MCP](https://modelcontextprotocol.io) (JSON or YAML) into a ready-to-run [MCP](https://modelcontextprotocol.io)
server using [FastMCP](https://gofastmcp.com). Each API operation becomes an MCP server using [FastMCP](https://gofastmcp.com). One MCP tool is generated per
tool; when several documents are configured they are mounted under a shared API operation, and requests to the target API are authenticated with a token
server and served on a single transport. supplied via environment variables.
All configuration lives in a YAML file (JSON is also accepted). Only the Everything is configured through environment variables: the spec location, the
optional path to that file can come from the environment (`XMCP_CONFIG`). transport (stdio or streamable HTTP), the listen host/port, and the API token.
## Quick start ## Quick start
### Docker Compose (HTTP)
```sh ```sh
cp xmcp.example.yml xmcp.yml # then edit specs[].spec_url and credentials cp .env.example .env # then edit OPENAPI_SPEC_URL / API_TOKEN etc.
docker compose up --build docker compose up --build
``` ```
The MCP endpoint is served at `http://localhost:8000/mcp`. The MCP endpoint is served at `http://localhost:8000/mcp`.
### Docker (without Compose) ### Docker (HTTP)
```sh ```sh
docker build -t xmcp . docker build -t xmcp .
docker run --rm -p 8000:8000 \ docker run --rm -p 8000:8000 \
-v "$PWD/xmcp.yml:/etc/xmcp/xmcp.yml:ro" \ -e OPENAPI_SPEC_URL=https://petstore3.swagger.io/api/v3/openapi.json \
xmcp xmcp
``` ```
For stdio, mount the config and run the container with an open stdin: ### Docker (stdio)
For stdio the container must run with an open stdin (`-i`); logs go to stderr,
so stdout stays protocol-clean:
```sh ```sh
docker run -i --rm \ docker run -i --rm \
-v "$PWD/xmcp.yml:/etc/xmcp/xmcp.yml:ro" \ -e MCP_TRANSPORT=stdio \
-e OPENAPI_SPEC_URL=https://petstore3.swagger.io/api/v3/openapi.json \
xmcp xmcp
``` ```
Example MCP client configuration for the stdio container:
```json
{
"mcpServers": {
"petstore": {
"command": "docker",
"args": [
"run", "-i", "--rm",
"-e", "MCP_TRANSPORT=stdio",
"-e", "OPENAPI_SPEC_URL=https://petstore3.swagger.io/api/v3/openapi.json",
"xmcp:latest"
]
}
}
}
```
### Local development with uv ### Local development with uv
```sh ```sh
uv sync uv sync
uv run xmcp --config xmcp.yml OPENAPI_SPEC_URL=https://petstore3.swagger.io/api/v3/openapi.json uv run xmcp
# equivalent: uv run python -m xmcp
``` ```
The configuration file is located, in order, via:
1. `--config PATH`
2. the `XMCP_CONFIG` environment variable
3. the default `/etc/xmcp/xmcp.yml`
## Configuration ## Configuration
```yaml | Variable | Required | Default | Description |
server: | --- | --- | --- | --- |
name: xmcp # optional; single-spec default is the spec title | `OPENAPI_SPEC_URL` | yes | – | `http(s)://` URL or local file path of the OpenAPI document (JSON or YAML) |
transport: http # stdio | http | `OPENAPI_SPEC_TOKEN` | no | – | Bearer token used only when fetching a remote spec document |
host: 0.0.0.0 | `API_BASE_URL` | no | spec `servers[0].url` | Overrides the target API base URL |
port: 8000 | `API_TOKEN` | no | – | Credential sent to the target API on every tool call |
log_level: INFO | `API_AUTH_HEADER` | no | `Authorization` | Header used to carry the token (e.g. `X-API-Key`) |
http_timeout: 30 | `API_AUTH_SCHEME` | no | `Bearer` | Scheme prefix; set to an empty value to send the raw token |
client_api_key_header: X-API-Key # "" disables per-request key forwarding | `MCP_API_KEY_HEADER` | no | `X-API-Key` | HTTP header treated as the per-request API key (empty disables the mapping) |
| `MCP_TRANSPORT` | no | `stdio` (`http` in the image) | `stdio` or `http` (`streamable-http` is an alias) |
| `MCP_HOST` | no | `0.0.0.0` | Listen host for the HTTP transport |
| `MCP_PORT` | no | `8000` | Listen port for the HTTP transport |
| `MCP_SERVER_NAME` | no | spec `info.title` | Name advertised by the MCP server |
| `MCP_TOOLS_INCLUDED` | no | – (all tools) | Comma-separated allowlist of tool names to expose |
| `MCP_TOOLS_TAGS_INCLUDED` | no | – (all tags) | Comma-separated allowlist of OpenAPI tags to expose |
| `HTTP_TIMEOUT` | no | `30` | Timeout in seconds for spec fetch and API calls |
| `LOG_LEVEL` | no | `INFO` | Python log level |
api: # defaults applied to every OpenAPI document Authentication examples: for a bearer token leave the defaults; for an API key
# base_url: https://api.example.com header use `API_AUTH_HEADER=X-API-Key` with `API_AUTH_SCHEME` set to an empty
# token: ${API_TOKEN} string. At startup the generated tool list (name, tags and description) is
auth_header: Authorization logged at INFO; set `LOG_LEVEL=WARNING` to suppress it.
auth_scheme: Bearer # "" sends the raw token
tools: # optional global filters ### Per-client API keys (HTTP transport)
include: [getPetById] # namespaced names in multi-spec mode
tags_include: [pet] # OpenAPI operation tags
specs:
- spec_url: https://petstore3.swagger.io/api/v3/openapi.json
api:
base_url: https://petstore3.swagger.io/api/v3
```
| Key | Description |
| --- | --- |
| `server.name` | Name advertised by the parent server (multi-spec default: `xmcp`) |
| `server.transport` | `stdio` or `http` (default `stdio`) |
| `server.host`, `server.port` | HTTP listen address (defaults `0.0.0.0`, `8000`) |
| `server.log_level` | Python log level (default `INFO`) |
| `server.http_timeout` | Timeout in seconds for spec fetch and API calls (default `30`) |
| `server.client_api_key_header` | Header carrying a per-request API key (default `X-API-Key`, empty disables) |
| `api.base_url`, `api.token` | Defaults for every spec's target API |
| `api.auth_header`, `api.auth_scheme` | Default credential header and scheme |
| `tools.include` | Allowlist of tool names (namespaced in multi-spec mode) |
| `tools.tags_include` | Allowlist of OpenAPI tags |
| `specs[].namespace` | Tool-name prefix; required when more than one spec is configured |
| `specs[].spec_url` | HTTP(S) URL or local file path of the document |
| `specs[].spec_token` | Bearer token used only to fetch the document |
| `specs[].server_name` | MCP server name for this document |
| `specs[].api.*` | Per-spec overrides of the `api` defaults |
Notes:
- `${VAR}` in any string value is replaced with the environment variable. A
missing variable is a startup error; the error names the exact key.
- Unset or empty `tools` filters expose everything. `tools.include` and
`tools.tags_include` combine with AND semantics; entries that match nothing
are reported with a startup warning.
- Unknown keys, invalid values, duplicate namespaces and empty `specs` fail
fast with a clear message.
- Relative local paths in `spec_url` resolve against the config file's
directory, which keeps mounted deployments predictable.
- The generated tool list (name, tags, description) is logged at startup at
INFO; set `server.log_level: WARNING` to suppress it.
## Multiple OpenAPI documents
Give each entry a unique namespace:
```yaml
tools:
include: [pet_getPetById, billing_listInvoices]
tags_include: [pet, billing]
specs:
- namespace: pet
spec_url: https://petstore3.swagger.io/api/v3/openapi.json
api:
base_url: https://petstore3.swagger.io/api/v3
token: ${PETSTORE_TOKEN}
- namespace: billing
spec_url: ./specs/billing.yaml
api:
base_url: https://billing.example.com
token: ${BILLING_TOKEN}
auth_header: X-API-Key
auth_scheme: ""
```
Tools are exposed as `pet_getPetById`, `billing_listInvoices`, and so on;
resources and prompts are namespaced too. A single document without a namespace
keeps its original, unprefixed tool names.
## Per-client API keys (HTTP transport)
When serving over HTTP, clients can send their own API key on each request When serving over HTTP, clients can send their own API key on each request
instead of using the configured `api.token`. The value is treated as a raw instead of sharing the server's `API_TOKEN`. The value is treated as a raw
token and forwarded using each spec's `api.auth_header` / `api.auth_scheme` token and forwarded to the target API using `API_AUTH_HEADER` /
(by default `Authorization: Bearer <key>`): `API_AUTH_SCHEME` (by default `Authorization: Bearer <key>`):
```sh ```sh
curl -X POST http://localhost:8000/mcp \ curl -X POST http://localhost:8000/mcp \
@@ -161,51 +117,68 @@ curl -X POST http://localhost:8000/mcp \
-d '{"jsonrpc":"2.0","id":1,"method":"initialize",...}' -d '{"jsonrpc":"2.0","id":1,"method":"initialize",...}'
``` ```
Precedence: the request header wins, then `api.token`, then no authentication. Precedence: the request header wins, then `API_TOKEN`, then no authentication.
The header name is `server.client_api_key_header`; an empty value disables the The header name is configurable with `MCP_API_KEY_HEADER`; set it to an empty
mapping. When mapped to a differently named target header, the source header is value to disable the mapping. When the key is mapped to a differently named
removed from the outgoing API request so the credential is not sent twice. In target header, the source header is removed from the outgoing API request so
multi-spec mode the same request header is used for every mounted API, each the credential is not sent twice.
formatting it according to its own settings.
Note: FastMCP independently passes other incoming HTTP headers through to the Note: FastMCP independently passes other incoming HTTP headers through to the
API request, except credential headers such as `Authorization` and `Cookie`. API request, except credential headers such as `Authorization` and `Cookie`.
MCP clients that support custom transport headers should send the key on every
request (at least on `tools/call`).
### Tool allowlists
By default every operation in the OpenAPI document is exposed as an MCP tool.
Two environment variables can restrict the exposed set; each accepts a
comma-separated list, and unset or empty disables that filter:
```sh
# Only these operations
MCP_TOOLS_INCLUDED=getPetById,findPetsByStatus
# Only operations tagged "pet" in the OpenAPI document
MCP_TOOLS_TAGS_INCLUDED=pet
```
When both are set, a tool must match both filters (name **and** tag). Entries
that do not match anything are ignored with a startup warning, and the active
selection is reported at startup
(`Tool filters active (names: ...; tags: ...): exposing N of M tool(s): ...`).
### Serving a local spec file
```sh
docker run -i --rm \
-v "$PWD/openapi.json:/spec/openapi.json:ro" \
-e OPENAPI_SPEC_URL=/spec/openapi.json \
xmcp
```
If the document has no absolute `servers[0].url` (for example it only contains
a relative path such as `/v1`), set `API_BASE_URL` to the API root explicitly.
### Base URL resolution
The base URL for generated tools is chosen as follows:
1. `API_BASE_URL`, if set.
2. The first entry of the spec's `servers` list.
3. If that server URL is relative (e.g. `/api/v3`) and the spec was fetched
over HTTP(S), it is resolved against the spec's origin.
4. Otherwise startup fails with a message asking for `API_BASE_URL`.
## Security notes ## Security notes
- The configuration file may contain API tokens. Keep it out of version - The API token is attached to every request the generated tools make to the
control (it is gitignored), mount it read-only, and restrict its permissions. target API. It is never logged.
- `${VAR}` interpolation keeps secrets out of the file (e.g. `token: ${API_TOKEN}`).
- The MCP HTTP endpoint (`/mcp`) itself is **not** authenticated. Run it on a - The MCP HTTP endpoint (`/mcp`) itself is **not** authenticated. Run it on a
private network or behind a reverse proxy / gateway that enforces private network or behind a reverse proxy / gateway that enforces
authentication, and terminate TLS there so client keys are not sent in clear authentication before exposing it publicly.
text. - Per-request API keys (the `X-API-Key` header by default) are forwarded only to
- Per-request API keys are forwarded only to the target API and are never the target API and are never logged. Terminate TLS in front of the MCP
logged. endpoint so client keys are not sent in clear text.
## Migrating from environment variables
Configuration is now file-based. Legacy environment variables are ignored (a
startup warning lists any that are set); only `XMCP_CONFIG` (the config file
path) is still read.
| Previous environment variable | Configuration key |
| --- | --- |
| `OPENAPI_SPEC_URL` | `specs[].spec_url` |
| `OPENAPI_SPEC_TOKEN` | `specs[].spec_token` |
| `API_BASE_URL` | `api.base_url` / `specs[].api.base_url` |
| `API_TOKEN` | `api.token` / `specs[].api.token` |
| `API_AUTH_HEADER` | `api.auth_header` |
| `API_AUTH_SCHEME` | `api.auth_scheme` |
| `MCP_API_KEY_HEADER` | `server.client_api_key_header` |
| `MCP_TRANSPORT` | `server.transport` |
| `MCP_HOST` | `server.host` |
| `MCP_PORT` | `server.port` |
| `MCP_SERVER_NAME` | `server.name` |
| `MCP_TOOLS_INCLUDED` | `tools.include` |
| `MCP_TOOLS_TAGS_INCLUDED` | `tools.tags_include` |
| `HTTP_TIMEOUT` | `server.http_timeout` |
| `LOG_LEVEL` | `server.log_level` |
## Releasing ## Releasing
@@ -228,12 +201,10 @@ development version (for example `0.2.1.dev3+gabc1234`).
## Project layout ## Project layout
``` ```
src/xmcp/config.py configuration loading and validation src/xmcp/config.py env var parsing and validation
src/xmcp/spec.py spec loading, base URL resolution, API client src/xmcp/spec.py spec loading, base URL resolution, API client
src/xmcp/auth.py per-request API key forwarding (HTTP transport) src/xmcp/auth.py per-request API key forwarding (HTTP transport)
src/xmcp/server.py server construction, mounting, lifecycle src/xmcp/server.py FastMCP server construction and entrypoint
src/xmcp/healthcheck.py container healthcheck
Dockerfile multi-stage build (uv builder -> alpine runtime, non-root) Dockerfile multi-stage build (uv builder -> alpine runtime, non-root)
.gitea/workflows/build.yaml release pipeline (tag push -> docker build & push) .gitea/workflows/build.yaml release pipeline (tag push -> docker build & push)
xmcp.example.yml annotated configuration example
``` ```
+11 -4
View File
@@ -3,9 +3,16 @@ services:
build: build:
context: . context: .
image: xmcp:latest image: xmcp:latest
volumes: env_file:
- ./xmcp.yml:/etc/xmcp/xmcp.yml:ro - .env
# The container port must match server.port in xmcp.yml. environment:
MCP_TRANSPORT: http
MCP_HOST: 0.0.0.0
MCP_PORT: "8000"
ports: ports:
- "8000:8000" - "${MCP_PORT:-8000}:${MCP_PORT:-8000}"
restart: unless-stopped restart: unless-stopped
# To serve a locally stored OpenAPI document, mount it and point
# OPENAPI_SPEC_URL at the in-container path:
# volumes:
# - ./openapi.json:/spec/openapi.json:ro
+101 -394
View File
@@ -1,20 +1,10 @@
"""File-based configuration for xmcp.""" """Environment-driven configuration for xmcp."""
from __future__ import annotations from __future__ import annotations
import logging import logging
import os import os
import re
from collections.abc import Mapping
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path
from typing import Any
from urllib.parse import urlparse
import yaml
DEFAULT_CONFIG_PATH = Path("/etc/xmcp/xmcp.yml")
CONFIG_PATH_ENV = "XMCP_CONFIG"
VALID_TRANSPORTS = ("stdio", "http") VALID_TRANSPORTS = ("stdio", "http")
_TRANSPORT_ALIASES = { _TRANSPORT_ALIASES = {
@@ -23,407 +13,124 @@ _TRANSPORT_ALIASES = {
"streamablehttp": "http", "streamablehttp": "http",
} }
_NAMESPACE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]*$")
_ENV_PATTERN = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
_SERVER_KEYS = {
"name",
"transport",
"host",
"port",
"log_level",
"http_timeout",
"client_api_key_header",
}
_API_KEYS = {"base_url", "token", "auth_header", "auth_scheme"}
_TOOLS_KEYS = {"include", "tags_include"}
_SPEC_KEYS = {"namespace", "spec_url", "spec_token", "server_name", "api"}
_TOP_LEVEL_KEYS = {"server", "api", "tools", "specs"}
LEGACY_ENV_VARS = (
"OPENAPI_SPEC_URL",
"OPENAPI_SPEC_TOKEN",
"API_BASE_URL",
"API_TOKEN",
"API_AUTH_HEADER",
"API_AUTH_SCHEME",
"MCP_API_KEY_HEADER",
"MCP_TRANSPORT",
"MCP_HOST",
"MCP_PORT",
"MCP_SERVER_NAME",
"MCP_TOOLS_INCLUDED",
"MCP_TOOLS_TAGS_INCLUDED",
"HTTP_TIMEOUT",
"LOG_LEVEL",
)
class ConfigError(ValueError): class ConfigError(ValueError):
"""Raised when the configuration file is invalid.""" """Raised when the environment configuration is invalid."""
@dataclass(frozen=True) def _get(name: str) -> str | None:
class SpecSource: """Read an environment variable, treating blank values as unset."""
"""One OpenAPI document and its per-spec overrides.""" value = os.environ.get(name, "").strip()
return value or None
spec_url: str
namespace: str | None = None def _get_int(name: str, default: int, *, minimum: int, maximum: int | None = None) -> int:
spec_token: str | None = None raw = _get(name)
server_name: str | None = None if raw is None:
api_base_url: str | None = None return default
api_token: str | None = None try:
api_auth_header: str | None = None value = int(raw)
api_auth_scheme: str | None = None except ValueError:
raise ConfigError(f"{name} must be an integer, got {raw!r}") from None
if value < minimum or (maximum is not None and value > maximum):
bounds = f"at least {minimum}" if maximum is None else f"between {minimum} and {maximum}"
raise ConfigError(f"{name} must be {bounds}, got {value}")
return value
def _get_float(name: str, default: float, *, minimum: float) -> float:
raw = _get(name)
if raw is None:
return default
try:
value = float(raw)
except ValueError:
raise ConfigError(f"{name} must be a number, got {raw!r}") from None
if value < minimum:
raise ConfigError(f"{name} must be at least {minimum}, got {value}")
return value
def _parse_set(name: str) -> frozenset[str] | None:
"""Parse a comma-separated environment variable into a set of names."""
raw = _get(name)
if raw is None:
return None
values = frozenset(value.strip() for value in raw.split(",") if value.strip())
return values or None
@dataclass(frozen=True) @dataclass(frozen=True)
class Config: class Config:
"""Resolved runtime configuration.""" """Resolved runtime configuration."""
path: Path spec_url: str
specs: tuple[SpecSource, ...] spec_token: str | None
transport: str = "stdio" api_base_url: str | None
host: str = "0.0.0.0" api_token: str | None
port: int = 8000 api_auth_header: str
server_name: str | None = None api_auth_scheme: str
timeout: float = 30.0 mcp_api_key_header: str | None
log_level: str = "INFO" tools_included: frozenset[str] | None
client_api_key_header: str | None = "X-API-Key" tools_tags_included: frozenset[str] | None
tools_include: frozenset[str] | None = None transport: str
tools_tags_include: frozenset[str] | None = None host: str
api_base_url: str | None = None port: int
api_token: str | None = None server_name: str | None
api_auth_header: str = "Authorization" timeout: float
api_auth_scheme: str = "Bearer" log_level: str
def api_for(self, spec: SpecSource) -> tuple[str | None, str | None, str, str]: @classmethod
"""Return (base_url, token, header, scheme) for a spec, applying defaults.""" def from_env(cls) -> "Config":
base_url = spec.api_base_url if spec.api_base_url is not None else self.api_base_url """Build a Config from environment variables, raising ConfigError."""
token = spec.api_token if spec.api_token is not None else self.api_token spec_url = _get("OPENAPI_SPEC_URL")
header = ( if spec_url is None:
spec.api_auth_header
if spec.api_auth_header is not None
else self.api_auth_header
)
scheme = (
spec.api_auth_scheme
if spec.api_auth_scheme is not None
else self.api_auth_scheme
)
return base_url, token, header, scheme
def resolve_config_path(cli_path: str | None = None) -> Path:
"""Resolve the configuration path: --config, then XMCP_CONFIG, then default."""
if cli_path:
return Path(cli_path).expanduser()
env_path = os.environ.get(CONFIG_PATH_ENV, "").strip()
if env_path:
return Path(env_path).expanduser()
return DEFAULT_CONFIG_PATH
def legacy_environment_variables() -> list[str]:
"""Return legacy env vars that are set but no longer used."""
return [name for name in LEGACY_ENV_VARS if os.environ.get(name, "").strip()]
def load_config(cli_path: str | None = None) -> Config:
"""Load and validate the configuration file."""
path = resolve_config_path(cli_path)
document = _read_document(path)
return _parse_config(document, path)
# --------------------------------------------------------------------------- #
# File loading and interpolation
# --------------------------------------------------------------------------- #
def _read_document(path: Path) -> dict[str, Any]:
try:
text = path.read_text(encoding="utf-8")
except FileNotFoundError as exc:
raise ConfigError(
f"configuration file not found: {path} "
f"(use --config, ${CONFIG_PATH_ENV}, or create the default file)"
) from exc
except OSError as exc:
raise ConfigError(f"could not read configuration file {path}: {exc}") from exc
try:
document = yaml.safe_load(text)
except yaml.YAMLError as exc:
raise ConfigError(f"invalid YAML/JSON in {path}: {exc}") from exc
if not isinstance(document, dict):
raise ConfigError(f"configuration file {path} must contain a mapping at the top level")
return _interpolate(document, "")
def _interpolate(value: Any, location: str) -> Any:
if isinstance(value, dict):
return {
key: _interpolate(item, f"{location}.{key}" if location else str(key))
for key, item in value.items()
}
if isinstance(value, list):
return [
_interpolate(item, f"{location}[{index}]")
for index, item in enumerate(value)
]
if isinstance(value, str):
def replace(match: re.Match[str]) -> str:
name = match.group(1)
if name not in os.environ:
where = location or "<root>"
raise ConfigError(
f"environment variable {name!r} referenced at {where} is not set"
)
return os.environ[name]
return _ENV_PATTERN.sub(replace, value)
return value
# --------------------------------------------------------------------------- #
# Document parsing
# --------------------------------------------------------------------------- #
def _parse_config(document: Mapping[str, Any], path: Path) -> Config:
_reject_unknown(document, _TOP_LEVEL_KEYS, "the top level")
server = _mapping(document.get("server"), "server")
_reject_unknown(server, _SERVER_KEYS, "server")
transport = (_string(server, "transport", "server", "stdio") or "stdio").lower()
transport = _TRANSPORT_ALIASES.get(transport, transport)
if transport not in VALID_TRANSPORTS:
raise ConfigError(
f"server.transport must be one of {'/'.join(VALID_TRANSPORTS)} "
f"(or 'streamable-http'), got {transport!r}"
)
log_level = (_string(server, "log_level", "server", "INFO") or "INFO").upper()
if log_level not in logging.getLevelNamesMapping():
raise ConfigError(f"server.log_level is not a valid log level: {log_level!r}")
port = _int_value(server.get("port"), "server.port", 8000, 1, 65535)
timeout = _float_value(server.get("http_timeout"), "server.http_timeout", 30.0, 0.1)
host = _string(server, "host", "server", "0.0.0.0") or "0.0.0.0"
server_name = _blank_to_none(_string(server, "name", "server"))
if "client_api_key_header" in server:
raw_header = _string(server, "client_api_key_header", "server")
header = raw_header.strip() if raw_header is not None else ""
client_api_key_header = header or None
else:
client_api_key_header = "X-API-Key"
api = _mapping(document.get("api"), "api")
_reject_unknown(api, _API_KEYS, "api")
api_base_url = _blank_to_none(_string(api, "base_url", "api"))
api_token = _blank_to_none(_string(api, "token", "api"))
api_auth_header = _blank_to_none(_string(api, "auth_header", "api")) or "Authorization"
if "auth_scheme" in api:
raw_scheme = _string(api, "auth_scheme", "api")
api_auth_scheme = raw_scheme.strip() if raw_scheme is not None else ""
else:
api_auth_scheme = "Bearer"
tools = _mapping(document.get("tools"), "tools")
_reject_unknown(tools, _TOOLS_KEYS, "tools")
tools_include = _name_set(tools.get("include"), "tools.include")
tools_tags_include = _name_set(tools.get("tags_include"), "tools.tags_include")
specs_raw = document.get("specs")
if specs_raw is None:
raise ConfigError("specs is required: add at least one OpenAPI document")
if not isinstance(specs_raw, list) or not specs_raw:
raise ConfigError("specs must be a non-empty list")
base_dir = path.parent.resolve()
specs: list[SpecSource] = []
seen_namespaces: set[str] = set()
for index, raw_spec in enumerate(specs_raw):
location = f"specs[{index}]"
spec_map = _mapping(raw_spec, location)
_reject_unknown(spec_map, _SPEC_KEYS, location)
spec_url = _string(spec_map, "spec_url", location)
if not spec_url:
raise ConfigError(f"{location}.spec_url is required")
namespace = _blank_to_none(_string(spec_map, "namespace", location))
if namespace is not None:
if not _NAMESPACE_RE.match(namespace):
raise ConfigError(
f"{location}.namespace must match [A-Za-z0-9][A-Za-z0-9_-]*, "
f"got {namespace!r}"
)
if namespace in seen_namespaces:
raise ConfigError(f"duplicate namespace {namespace!r} in {location}")
seen_namespaces.add(namespace)
spec_api = _mapping(spec_map.get("api"), f"{location}.api")
_reject_unknown(spec_api, _API_KEYS, f"{location}.api")
spec_api_scheme: str | None = None
if "auth_scheme" in spec_api:
raw_spec_scheme = _string(spec_api, "auth_scheme", f"{location}.api")
spec_api_scheme = (
raw_spec_scheme.strip() if raw_spec_scheme is not None else ""
)
specs.append(
SpecSource(
spec_url=_resolve_spec_location(spec_url, base_dir),
namespace=namespace,
spec_token=_blank_to_none(_string(spec_map, "spec_token", location)),
server_name=_blank_to_none(_string(spec_map, "server_name", location)),
api_base_url=_blank_to_none(
_string(spec_api, "base_url", f"{location}.api")
),
api_token=_blank_to_none(_string(spec_api, "token", f"{location}.api")),
api_auth_header=_blank_to_none(
_string(spec_api, "auth_header", f"{location}.api")
),
api_auth_scheme=spec_api_scheme,
)
)
if len(specs) > 1:
missing = [
f"specs[{index}]"
for index, spec in enumerate(specs)
if spec.namespace is None
]
if missing:
raise ConfigError( raise ConfigError(
"namespace is required for every entry when multiple specs are " "OPENAPI_SPEC_URL is required: set it to an http(s) URL or a local "
f"configured (missing in {', '.join(missing)})" "path of an OpenAPI document (JSON or YAML)"
) )
return Config( transport = (_get("MCP_TRANSPORT") or "stdio").lower()
path=path.resolve(), transport = _TRANSPORT_ALIASES.get(transport, transport)
specs=tuple(specs), if transport not in VALID_TRANSPORTS:
transport=transport, raise ConfigError(
host=host, f"MCP_TRANSPORT must be one of {'/'.join(VALID_TRANSPORTS)} "
port=port, f"(or 'streamable-http'), got {transport!r}"
server_name=server_name, )
timeout=timeout,
log_level=log_level,
client_api_key_header=client_api_key_header,
tools_include=tools_include,
tools_tags_include=tools_tags_include,
api_base_url=api_base_url,
api_token=api_token,
api_auth_header=api_auth_header,
api_auth_scheme=api_auth_scheme,
)
log_level = (_get("LOG_LEVEL") or "INFO").upper()
if log_level not in logging.getLevelNamesMapping():
raise ConfigError(f"LOG_LEVEL is not a valid log level: {log_level!r}")
# --------------------------------------------------------------------------- # # Read without the blank-means-unset helper on purpose: an explicitly
# Small parsing helpers # empty value means "send the raw token with no scheme prefix".
# --------------------------------------------------------------------------- # raw_scheme = os.environ.get("API_AUTH_SCHEME")
auth_scheme = "Bearer" if raw_scheme is None else raw_scheme.strip()
# Explicitly empty disables forwarding of per-request API keys.
raw_key_header = os.environ.get("MCP_API_KEY_HEADER")
key_header = "X-API-Key" if raw_key_header is None else raw_key_header.strip()
key_header = key_header or None
def _mapping(value: Any, location: str) -> Mapping[str, Any]: # Comma-separated allowlists of tool names / OpenAPI tags; unset or
if value is None: # empty means no filtering on that dimension.
return {} included = _parse_set("MCP_TOOLS_INCLUDED")
if not isinstance(value, dict): tags_included = _parse_set("MCP_TOOLS_TAGS_INCLUDED")
raise ConfigError(f"{location} must be a mapping")
return value
return cls(
def _reject_unknown( spec_url=spec_url,
mapping: Mapping[str, Any], allowed: set[str], location: str spec_token=_get("OPENAPI_SPEC_TOKEN"),
) -> None: api_base_url=_get("API_BASE_URL"),
unknown = sorted(str(key) for key in set(mapping) - allowed) api_token=_get("API_TOKEN"),
if unknown: api_auth_header=_get("API_AUTH_HEADER") or "Authorization",
raise ConfigError(f"unknown key(s) in {location}: {', '.join(unknown)}") api_auth_scheme=auth_scheme,
mcp_api_key_header=key_header,
tools_included=included,
def _string( tools_tags_included=tags_included,
mapping: Mapping[str, Any], transport=transport,
key: str, host=_get("MCP_HOST") or "0.0.0.0",
location: str, port=_get_int("MCP_PORT", 8000, minimum=1, maximum=65535),
default: str | None = None, server_name=_get("MCP_SERVER_NAME"),
) -> str | None: timeout=_get_float("HTTP_TIMEOUT", 30.0, minimum=0.1),
if key not in mapping or mapping[key] is None: log_level=log_level,
return default
value = mapping[key]
if not isinstance(value, str):
raise ConfigError(f"{location}.{key} must be a string")
return value
def _blank_to_none(value: str | None) -> str | None:
if value is None:
return None
value = value.strip()
return value or None
def _int_value(
value: Any, location: str, default: int, minimum: int, maximum: int
) -> int:
if value is None:
return default
if isinstance(value, bool):
raise ConfigError(f"{location} must be an integer")
try:
parsed = int(value)
except (TypeError, ValueError):
raise ConfigError(f"{location} must be an integer, got {value!r}") from None
if not minimum <= parsed <= maximum:
raise ConfigError(
f"{location} must be between {minimum} and {maximum}, got {parsed}"
) )
return parsed
def _float_value(
value: Any, location: str, default: float, minimum: float
) -> float:
if value is None:
return default
if isinstance(value, bool):
raise ConfigError(f"{location} must be a number")
try:
parsed = float(value)
except (TypeError, ValueError):
raise ConfigError(f"{location} must be a number, got {value!r}") from None
if parsed < minimum:
raise ConfigError(f"{location} must be at least {minimum}, got {parsed}")
return parsed
def _name_set(value: Any, location: str) -> frozenset[str] | None:
if value is None:
return None
if isinstance(value, str):
items = value.split(",")
elif isinstance(value, list):
if not all(isinstance(item, str) for item in value):
raise ConfigError(f"{location} must be a list of strings")
items = value
else:
raise ConfigError(
f"{location} must be a list of strings or a comma-separated string"
)
names = frozenset(item.strip() for item in items if item.strip())
return names or None
def _resolve_spec_location(raw: str, base_dir: Path) -> str:
"""Resolve local spec paths relative to the configuration file."""
scheme = urlparse(raw).scheme.lower()
if scheme in ("http", "https", "file"):
return raw
candidate = Path(raw).expanduser()
if not candidate.is_absolute():
candidate = base_dir / candidate
return str(candidate)
-31
View File
@@ -1,31 +0,0 @@
"""Container healthcheck: probe the MCP HTTP port when that transport is configured."""
from __future__ import annotations
import socket
import sys
from .config import ConfigError, load_config
def main() -> None:
try:
config = load_config()
except ConfigError as exc:
print(f"healthcheck: configuration error: {exc}", file=sys.stderr)
raise SystemExit(1) from exc
if config.transport != "http":
return
host = "127.0.0.1" if config.host in ("0.0.0.0", "::") else config.host
try:
with socket.create_connection((host, config.port), timeout=3):
return
except OSError as exc:
print(f"healthcheck: {exc}", file=sys.stderr)
raise SystemExit(1) from exc
if __name__ == "__main__":
main()
+40 -110
View File
@@ -1,91 +1,59 @@
"""Build and run the FastMCP server(s) generated from OpenAPI documents.""" """Build and run the FastMCP server generated from the OpenAPI document."""
from __future__ import annotations from __future__ import annotations
import argparse
import asyncio import asyncio
import logging import logging
import os import os
from fastmcp import FastMCP from fastmcp import FastMCP
from .config import ( from .config import Config, ConfigError
CONFIG_PATH_ENV,
DEFAULT_CONFIG_PATH,
Config,
ConfigError,
SpecSource,
legacy_environment_variables,
load_config,
)
from .spec import SpecError, build_api_client, load_spec, resolve_base_url from .spec import SpecError, build_api_client, load_spec, resolve_base_url
logger = logging.getLogger("xmcp") logger = logging.getLogger("xmcp")
def _build_subserver( def build_server(config: Config) -> FastMCP:
config: Config, spec: SpecSource, default_name: str | None """Load the OpenAPI document and build the FastMCP server."""
) -> FastMCP: spec = load_spec(config.spec_url, token=config.spec_token, timeout=config.timeout)
"""Build one FastMCP server from a single OpenAPI document.""" base_url = resolve_base_url(spec, override=config.api_base_url, spec_url=config.spec_url)
document = load_spec(spec.spec_url, token=spec.spec_token, timeout=config.timeout)
api_base_url, api_token, api_auth_header, api_auth_scheme = config.api_for(spec)
base_url = resolve_base_url(
document, override=api_base_url, spec_url=spec.spec_url
)
client = build_api_client( client = build_api_client(
base_url=base_url, base_url=base_url,
token=api_token, token=config.api_token,
header=api_auth_header, header=config.api_auth_header,
scheme=api_auth_scheme, scheme=config.api_auth_scheme,
timeout=config.timeout, timeout=config.timeout,
forwarded_header=config.client_api_key_header, forwarded_header=config.mcp_api_key_header,
) )
info = document.get("info") if isinstance(document.get("info"), dict) else {} info = spec.get("info") if isinstance(spec.get("info"), dict) else {}
name = ( name = config.server_name or str(info.get("title") or "OpenAPI MCP Server")
spec.server_name
or default_name
or str(info.get("title") or "OpenAPI MCP Server")
)
server = FastMCP.from_openapi(openapi_spec=document, client=client, name=name) server = FastMCP.from_openapi(openapi_spec=spec, client=client, name=name)
logger.info( logger.info(
"Generated MCP server %r from %s: %d path(s), target API %s", "Generated MCP server %r from %s: %d path(s), target API %s",
name, name,
spec.spec_url, config.spec_url,
len(document.get("paths") or {}), len(spec.get("paths") or {}),
base_url, base_url,
) )
if api_token is not None: if config.api_token:
logger.debug("API token configured for %r (%s header)", name, api_auth_header) logger.debug(
"Fallback API token configured for the %s header (%s)",
config.api_auth_header,
config.api_auth_scheme or "raw token",
)
if config.mcp_api_key_header:
logger.debug(
"Per-request API keys accepted from the %s header",
config.mcp_api_key_header,
)
elif not config.api_token:
logger.debug("No API token configured: requests are sent unauthenticated")
return server return server
def build_server(config: Config) -> FastMCP:
"""Build the MCP server from the configured OpenAPI documents."""
single = len(config.specs) == 1 and config.specs[0].namespace is None
default_name = config.server_name if single else None
subservers = [
(spec, _build_subserver(config, spec, default_name))
for spec in config.specs
]
if single:
return subservers[0][1]
parent = FastMCP(config.server_name or "xmcp")
for spec, subserver in subservers:
parent.mount(subserver, namespace=spec.namespace)
logger.info(
"Mounted %d OpenAPI server(s) under %r: %s",
len(subservers),
parent.name,
", ".join(f"{spec.namespace} <- {spec.spec_url}" for spec, _ in subservers),
)
return parent
async def _prepare_tools( async def _prepare_tools(
server: FastMCP, server: FastMCP,
included: frozenset[str] | None, included: frozenset[str] | None,
@@ -103,7 +71,7 @@ async def _prepare_tools(
unknown = sorted(included - known_names) unknown = sorted(included - known_names)
if unknown: if unknown:
logger.warning( logger.warning(
"tools.include lists unknown tool(s): %s", "MCP_TOOLS_INCLUDED lists unknown tool(s): %s",
", ".join(unknown), ", ".join(unknown),
) )
names = set(included & known_names) names = set(included & known_names)
@@ -113,7 +81,7 @@ async def _prepare_tools(
unknown_tags = sorted(tags_included - known_tags) unknown_tags = sorted(tags_included - known_tags)
if unknown_tags: if unknown_tags:
logger.warning( logger.warning(
"tools.tags_include lists unknown tag(s): %s", "MCP_TOOLS_TAGS_INCLUDED lists unknown tag(s): %s",
", ".join(unknown_tags), ", ".join(unknown_tags),
) )
tags = set(tags_included & known_tags) tags = set(tags_included & known_tags)
@@ -152,27 +120,9 @@ async def _prepare_tools(
logger.info(" - %s%s%s", tool.name, tags, suffix) logger.info(" - %s%s%s", tool.name, tags, suffix)
def _parse_args(argv: list[str] | None) -> argparse.Namespace: def main() -> None:
parser = argparse.ArgumentParser(
prog="xmcp",
description="Expose one or more OpenAPI documents as MCP servers.",
)
parser.add_argument(
"-c",
"--config",
metavar="PATH",
help=(
f"Configuration file (default: ${CONFIG_PATH_ENV} "
f"or {DEFAULT_CONFIG_PATH})"
),
)
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> None:
args = _parse_args(argv)
try: try:
config = load_config(args.config) config = Config.from_env()
except ConfigError as exc: except ConfigError as exc:
raise SystemExit(f"Configuration error: {exc}") from exc raise SystemExit(f"Configuration error: {exc}") from exc
@@ -180,49 +130,29 @@ def main(argv: list[str] | None = None) -> None:
level=config.log_level, level=config.log_level,
format="%(asctime)s %(levelname)-8s %(name)s: %(message)s", format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
) )
# Keep FastMCP's own logging in sync with the configured level. # Keep FastMCP's own logging in sync with LOG_LEVEL unless set explicitly.
os.environ.setdefault("FASTMCP_LOG_LEVEL", config.log_level) os.environ.setdefault("FASTMCP_LOG_LEVEL", config.log_level)
legacy = legacy_environment_variables()
if legacy:
logger.warning(
"Ignoring legacy environment variable(s): %s "
"(configuration is file-based: %s)",
", ".join(legacy),
config.path,
)
if config.client_api_key_header:
logger.debug(
"Per-request API keys accepted from the %s header",
config.client_api_key_header,
)
try: try:
server = build_server(config) server = build_server(config)
except SpecError as exc: except SpecError as exc:
raise SystemExit(f"Failed to load an OpenAPI document: {exc}") from exc raise SystemExit(f"Failed to load the OpenAPI document: {exc}") from exc
if ( if (
config.tools_include is not None config.tools_included is not None
or config.tools_tags_include is not None or config.tools_tags_included is not None
or logger.isEnabledFor(logging.INFO) or logger.isEnabledFor(logging.INFO)
): ):
asyncio.run( asyncio.run(
_prepare_tools(server, config.tools_include, config.tools_tags_include) _prepare_tools(
server, config.tools_included, config.tools_tags_included
)
) )
# Logs go to stderr so the stdio transport's stdout stays protocol-clean. # Logs go to stderr so the stdio transport's stdout stays protocol-clean.
if config.transport == "http": if config.transport == "http":
logger.info( logger.info("Serving MCP over streamable HTTP at http://%s:%d/mcp", config.host, config.port)
"Serving MCP over streamable HTTP at http://%s:%d/mcp",
config.host,
config.port,
)
server.run(transport="http", host=config.host, port=config.port) server.run(transport="http", host=config.host, port=config.port)
else: else:
logger.info("Serving MCP over stdio") logger.info("Serving MCP over stdio")
server.run(transport="stdio") server.run(transport="stdio")
if __name__ == "__main__":
main()
-47
View File
@@ -1,47 +0,0 @@
# xmcp configuration. Copy to xmcp.yml and adjust.
# Settings are read from this file only; ${VAR} placeholders are filled from
# the environment (a missing variable is a startup error).
server:
name: xmcp
transport: http # stdio | http
host: 0.0.0.0
port: 8000
log_level: INFO
http_timeout: 30
# Header from which MCP clients may supply a per-request API key.
# Empty string disables forwarding.
client_api_key_header: X-API-Key
# Defaults applied to every OpenAPI document (per-spec `api` keys override).
api:
# base_url: https://api.example.com
# token: ${API_TOKEN}
auth_header: Authorization
auth_scheme: Bearer # "" sends the raw token
# Optional global tool filters.
# tools:
# include: [getPetById, findPetsByStatus]
# tags_include: [pet]
specs:
# A single document needs no namespace; tool names stay unprefixed.
- spec_url: https://petstore3.swagger.io/api/v3/openapi.json
api:
base_url: https://petstore3.swagger.io/api/v3
# Multiple documents: give every entry a unique namespace. Tools are then
# exposed as "<namespace>_<tool>".
# specs:
# - namespace: pet
# spec_url: https://petstore3.swagger.io/api/v3/openapi.json
# api:
# base_url: https://petstore3.swagger.io/api/v3
# - namespace: billing
# spec_url: ./specs/billing.yaml
# api:
# base_url: https://billing.example.com
# token: ${BILLING_TOKEN}
# auth_header: X-API-Key
# auth_scheme: ""