Initial commit: xmcp — OpenAPI-to-MCP server

MCP server generated from an OpenAPI document with FastMCP v4.
Configurable via env vars (spec URL/file, transport stdio/http,
host, port, API token), built with uv, multi-stage Dockerfile
with non-root runtime.

.dockerignore

@@ -0,0 +1,16 @@
+.git/
+.gitignore
+.venv/
+__pycache__/
+*.py[cod]
+.pytest_cache/
+.ruff_cache/
+.mypy_cache/
+.env
+.env.*
+!.env.example
+docker-compose*.yml
+Dockerfile
+.dockerignore
+*.md
+!README.md

.env.example

@@ -0,0 +1,35 @@
+# 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=
+
+# --- 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
+
+# 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

.gitignore

@@ -0,0 +1,11 @@
+# Python
+__pycache__/
+*.py[cod]
+
+# uv / virtual environments
+.venv/
+
+# Local configuration
+.env
+.env.*
+!.env.example

.python-version

@@ -0,0 +1 @@
+3.13

Dockerfile

@@ -0,0 +1,49 @@
+# syntax=docker/dockerfile:1
+
+# ---- Build stage -------------------------------------------------------------
+# Uses the uv image to resolve and install the pinned dependency set into a
+# self-contained virtual environment.
+FROM ghcr.io/astral-sh/uv:python3.13-bookworm-slim AS builder
+
+ENV UV_COMPILE_BYTECODE=1 \
+    UV_LINK_MODE=copy \
+    UV_PYTHON_DOWNLOADS=never
+
+WORKDIR /app
+
+# Dependencies first: this layer only rebuilds when pyproject.toml/uv.lock change.
+COPY pyproject.toml uv.lock .python-version ./
+RUN --mount=type=cache,target=/root/.cache/uv \
+    uv sync --frozen --no-dev --no-install-project
+
+# Then install the project itself as a regular wheel.
+COPY README.md ./
+COPY src ./src
+RUN --mount=type=cache,target=/root/.cache/uv \
+    uv sync --frozen --no-dev --no-editable
+
+# ---- Runtime stage -----------------------------------------------------------
+FROM python:3.13-slim-bookworm AS runtime
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PATH="/app/.venv/bin:$PATH" \
+    MCP_TRANSPORT=http \
+    MCP_HOST=0.0.0.0 \
+    MCP_PORT=8000
+
+RUN groupadd --system --gid 1001 xmcp \
+ && useradd --system --uid 1001 --gid xmcp xmcp
+
+WORKDIR /app
+COPY --from=builder --chown=xmcp:xmcp /app/.venv /app/.venv
+
+USER xmcp
+
+EXPOSE 8000
+
+# In HTTP mode, probe the MCP port; stdio containers are always healthy.
+HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
+    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"]

README.md

@@ -0,0 +1,128 @@
+# xmcp
+
+Turns any [OpenAPI](https://spec.openapis.org/oas/latest.html) document
+(JSON or YAML) into a ready-to-run [MCP](https://modelcontextprotocol.io)
+server using [FastMCP](https://gofastmcp.com). One MCP tool is generated per
+API operation, and requests to the target API are authenticated with a token
+supplied via environment variables.
+
+Everything is configured through environment variables: the spec location, the
+transport (stdio or streamable HTTP), the listen host/port, and the API token.
+
+## Quick start
+
+### Docker Compose (HTTP)
+
+```sh
+cp .env.example .env   # then edit OPENAPI_SPEC_URL / API_TOKEN etc.
+docker compose up --build
+```
+
+The MCP endpoint is served at `http://localhost:8000/mcp`.
+
+### Docker (HTTP)
+
+```sh
+docker build -t xmcp .
+docker run --rm -p 8000:8000 \
+  -e OPENAPI_SPEC_URL=https://petstore3.swagger.io/api/v3/openapi.json \
+  xmcp
+```
+
+### Docker (stdio)
+
+For stdio the container must run with an open stdin (`-i`); logs go to stderr,
+so stdout stays protocol-clean:
+
+```sh
+docker run -i --rm \
+  -e MCP_TRANSPORT=stdio \
+  -e OPENAPI_SPEC_URL=https://petstore3.swagger.io/api/v3/openapi.json \
+  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
+
+```sh
+uv sync
+OPENAPI_SPEC_URL=https://petstore3.swagger.io/api/v3/openapi.json uv run xmcp
+# equivalent: uv run python -m xmcp
+```
+
+## Configuration
+
+| Variable | Required | Default | Description |
+| --- | --- | --- | --- |
+| `OPENAPI_SPEC_URL` | yes | – | `http(s)://` URL or local file path of the OpenAPI document (JSON or YAML) |
+| `OPENAPI_SPEC_TOKEN` | no | – | Bearer token used only when fetching a remote spec document |
+| `API_BASE_URL` | no | spec `servers[0].url` | Overrides the target API base URL |
+| `API_TOKEN` | no | – | Credential sent to the target API on every tool call |
+| `API_AUTH_HEADER` | no | `Authorization` | Header used to carry the token (e.g. `X-API-Key`) |
+| `API_AUTH_SCHEME` | no | `Bearer` | Scheme prefix; set to an empty value to send the raw token |
+| `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 |
+| `HTTP_TIMEOUT` | no | `30` | Timeout in seconds for spec fetch and API calls |
+| `LOG_LEVEL` | no | `INFO` | Python log level |
+
+Authentication examples: for a bearer token leave the defaults; for an API key
+header use `API_AUTH_HEADER=X-API-Key` with `API_AUTH_SCHEME` set to an empty
+string.
+
+### 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
+
+- The API token is attached to every request the generated tools make to the
+  target API. It is never logged.
+- The MCP HTTP endpoint (`/mcp`) itself is **not** authenticated. Run it on a
+  private network or behind a reverse proxy / gateway that enforces
+  authentication before exposing it publicly.
+
+## Project layout
+
+```
+src/xmcp/config.py   env var parsing and validation
+src/xmcp/spec.py     spec loading, base URL resolution, API client
+src/xmcp/server.py   FastMCP server construction and entrypoint
+Dockerfile           multi-stage build (uv builder -> slim runtime, non-root)
+```

docker-compose.yml

@@ -0,0 +1,18 @@
+services:
+  xmcp:
+    build:
+      context: .
+    image: xmcp:latest
+    env_file:
+      - .env
+    environment:
+      MCP_TRANSPORT: http
+      MCP_HOST: 0.0.0.0
+      MCP_PORT: "8000"
+    ports:
+      - "${MCP_PORT:-8000}:${MCP_PORT:-8000}"
+    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

pyproject.toml

@@ -0,0 +1,21 @@
+[project]
+name = "xmcp"
+version = "0.1.0"
+description = "MCP server generated from an OpenAPI specification, built with FastMCP"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "fastmcp>=4.1,<5",
+    "httpx2>=2.5",
+    "pyyaml>=6.0",
+]
+
+[project.scripts]
+xmcp = "xmcp.server:main"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/xmcp"]

src/xmcp/__init__.py

@@ -0,0 +1,3 @@
+"""xmcp: an MCP server generated from an OpenAPI specification with FastMCP."""
+
+__version__ = "0.1.0"

src/xmcp/__main__.py

@@ -0,0 +1,6 @@
+"""Entrypoint for ``python -m xmcp``."""
+
+from .server import main
+
+if __name__ == "__main__":
+    main()

src/xmcp/config.py

@@ -0,0 +1,111 @@
+"""Environment-driven configuration for xmcp."""
+
+from __future__ import annotations
+
+import logging
+import os
+from dataclasses import dataclass
+
+VALID_TRANSPORTS = ("stdio", "http")
+_TRANSPORT_ALIASES = {
+    "streamable-http": "http",
+    "streamable_http": "http",
+    "streamablehttp": "http",
+}
+
+
+class ConfigError(ValueError):
+    """Raised when the environment configuration is invalid."""
+
+
+def _get(name: str) -> str | None:
+    """Read an environment variable, treating blank values as unset."""
+    value = os.environ.get(name, "").strip()
+    return value or None
+
+
+def _get_int(name: str, default: int, *, minimum: int, maximum: int | None = None) -> int:
+    raw = _get(name)
+    if raw is None:
+        return default
+    try:
+        value = int(raw)
+    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
+
+
+@dataclass(frozen=True)
+class Config:
+    """Resolved runtime configuration."""
+
+    spec_url: str
+    spec_token: str | None
+    api_base_url: str | None
+    api_token: str | None
+    api_auth_header: str
+    api_auth_scheme: str
+    transport: str
+    host: str
+    port: int
+    server_name: str | None
+    timeout: float
+    log_level: str
+
+    @classmethod
+    def from_env(cls) -> "Config":
+        """Build a Config from environment variables, raising ConfigError."""
+        spec_url = _get("OPENAPI_SPEC_URL")
+        if spec_url is None:
+            raise ConfigError(
+                "OPENAPI_SPEC_URL is required: set it to an http(s) URL or a local "
+                "path of an OpenAPI document (JSON or YAML)"
+            )
+
+        transport = (_get("MCP_TRANSPORT") or "stdio").lower()
+        transport = _TRANSPORT_ALIASES.get(transport, transport)
+        if transport not in VALID_TRANSPORTS:
+            raise ConfigError(
+                f"MCP_TRANSPORT must be one of {'/'.join(VALID_TRANSPORTS)} "
+                f"(or 'streamable-http'), got {transport!r}"
+            )
+
+        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
+        # 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()
+
+        return cls(
+            spec_url=spec_url,
+            spec_token=_get("OPENAPI_SPEC_TOKEN"),
+            api_base_url=_get("API_BASE_URL"),
+            api_token=_get("API_TOKEN"),
+            api_auth_header=_get("API_AUTH_HEADER") or "Authorization",
+            api_auth_scheme=auth_scheme,
+            transport=transport,
+            host=_get("MCP_HOST") or "0.0.0.0",
+            port=_get_int("MCP_PORT", 8000, minimum=1, maximum=65535),
+            server_name=_get("MCP_SERVER_NAME"),
+            timeout=_get_float("HTTP_TIMEOUT", 30.0, minimum=0.1),
+            log_level=log_level,
+        )

src/xmcp/server.py

@@ -0,0 +1,74 @@
+"""Build and run the FastMCP server generated from the OpenAPI document."""
+
+from __future__ import annotations
+
+import logging
+import os
+
+from fastmcp import FastMCP
+
+from .config import Config, ConfigError
+from .spec import SpecError, build_api_client, load_spec, resolve_base_url
+
+logger = logging.getLogger("xmcp")
+
+
+def build_server(config: Config) -> FastMCP:
+    """Load the OpenAPI document and build the FastMCP server."""
+    spec = load_spec(config.spec_url, token=config.spec_token, timeout=config.timeout)
+    base_url = resolve_base_url(spec, override=config.api_base_url, spec_url=config.spec_url)
+    client = build_api_client(
+        base_url=base_url,
+        token=config.api_token,
+        header=config.api_auth_header,
+        scheme=config.api_auth_scheme,
+        timeout=config.timeout,
+    )
+
+    info = spec.get("info") if isinstance(spec.get("info"), dict) else {}
+    name = config.server_name or str(info.get("title") or "OpenAPI MCP Server")
+
+    server = FastMCP.from_openapi(openapi_spec=spec, client=client, name=name)
+    logger.info(
+        "Generated MCP server %r from %s: %d path(s), target API %s",
+        name,
+        config.spec_url,
+        len(spec.get("paths") or {}),
+        base_url,
+    )
+    if config.api_token:
+        logger.debug(
+            "API requests carry the %s header (%s)",
+            config.api_auth_header,
+            config.api_auth_scheme or "raw token",
+        )
+    else:
+        logger.debug("No API token configured: requests are sent unauthenticated")
+    return server
+
+
+def main() -> None:
+    try:
+        config = Config.from_env()
+    except ConfigError as exc:
+        raise SystemExit(f"Configuration error: {exc}") from exc
+
+    logging.basicConfig(
+        level=config.log_level,
+        format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
+    )
+    # Keep FastMCP's own logging in sync with LOG_LEVEL unless set explicitly.
+    os.environ.setdefault("FASTMCP_LOG_LEVEL", config.log_level)
+
+    try:
+        server = build_server(config)
+    except SpecError as exc:
+        raise SystemExit(f"Failed to load the OpenAPI document: {exc}") from exc
+
+    # Logs go to stderr so the stdio transport's stdout stays protocol-clean.
+    if config.transport == "http":
+        logger.info("Serving MCP over streamable HTTP at http://%s:%d/mcp", config.host, config.port)
+        server.run(transport="http", host=config.host, port=config.port)
+    else:
+        logger.info("Serving MCP over stdio")
+        server.run(transport="stdio")

src/xmcp/spec.py

@@ -0,0 +1,136 @@
+"""Loading of the OpenAPI document and construction of the target API client."""
+
+from __future__ import annotations
+
+import json
+import logging
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlparse
+from urllib.request import url2pathname
+
+import httpx2
+import yaml
+
+logger = logging.getLogger("xmcp.spec")
+
+_REMOTE_SCHEMES = ("http", "https")
+_SPEC_ACCEPT = "application/json, application/yaml, text/yaml, */*;q=0.1"
+
+
+class SpecError(RuntimeError):
+    """Raised when the OpenAPI document cannot be loaded."""
+
+
+def is_remote_spec(spec_url: str) -> bool:
+    """Whether the configured spec location is fetched over HTTP(S)."""
+    return urlparse(spec_url).scheme.lower() in _REMOTE_SCHEMES
+
+
+def load_spec(spec_url: str, *, token: str | None, timeout: float) -> dict[str, Any]:
+    """Load and parse the OpenAPI document from an HTTP(S) URL or a file path."""
+    scheme = urlparse(spec_url).scheme.lower()
+    if scheme in _REMOTE_SCHEMES:
+        raw = _fetch_remote(spec_url, token=token, timeout=timeout)
+    elif scheme == "file":
+        raw = _read_file(Path(url2pathname(urlparse(spec_url).path)))
+    elif scheme == "":
+        raw = _read_file(Path(spec_url).expanduser())
+    else:
+        raise SpecError(
+            f"Unsupported scheme {scheme!r} in OPENAPI_SPEC_URL: "
+            "use an http(s) URL or a local file path"
+        )
+    return _parse(raw, source=spec_url)
+
+
+def resolve_base_url(spec: Mapping[str, Any], *, override: str | None, spec_url: str) -> str:
+    """Determine the base URL that generated tools will call."""
+    if override:
+        return override.rstrip("/")
+
+    servers = spec.get("servers") or []
+    server_url = ""
+    if servers and isinstance(servers[0], Mapping):
+        server_url = str(servers[0].get("url") or "").strip()
+
+    if server_url:
+        if urlparse(server_url).scheme:
+            return server_url.rstrip("/")
+        # A relative server URL (e.g. "/api/v1") is resolved against the origin
+        # the spec was fetched from, which is the usual hosting convention.
+        if is_remote_spec(spec_url):
+            parsed = urlparse(spec_url)
+            origin = f"{parsed.scheme}://{parsed.netloc}"
+            return f"{origin}/{server_url.lstrip('/')}".rstrip("/")
+
+    raise SpecError(
+        "Could not determine the API base URL from the OpenAPI document; "
+        "set API_BASE_URL to the target API root (e.g. https://api.example.com)"
+    )
+
+
+def build_api_client(
+    *,
+    base_url: str,
+    token: str | None,
+    header: str,
+    scheme: str,
+    timeout: float,
+) -> httpx2.AsyncClient:
+    """Create the HTTP client used by the generated tools."""
+    headers: dict[str, str] = {}
+    if token:
+        headers[header] = f"{scheme} {token}".strip() if scheme else token
+    return httpx2.AsyncClient(base_url=base_url, headers=headers, timeout=timeout)
+
+
+def _fetch_remote(url: str, *, token: str | None, timeout: float) -> bytes:
+    headers = {"Accept": _SPEC_ACCEPT}
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    try:
+        with httpx2.Client(timeout=timeout, follow_redirects=True) as client:
+            response = client.get(url, headers=headers)
+            response.raise_for_status()
+            return response.content
+    except httpx2.HTTPError as exc:
+        raise SpecError(f"Failed to fetch the OpenAPI document from {url}: {exc}") from exc
+
+
+def _read_file(path: Path) -> bytes:
+    resolved = path.resolve()
+    if not resolved.exists():
+        raise SpecError(f"OpenAPI document not found: {resolved}")
+    if not resolved.is_file():
+        raise SpecError(f"OPENAPI_SPEC_URL does not point to a file: {resolved}")
+    try:
+        return resolved.read_bytes()
+    except OSError as exc:
+        raise SpecError(f"Could not read the OpenAPI document at {resolved}: {exc}") from exc
+
+
+def _parse(raw: bytes, *, source: str) -> dict[str, Any]:
+    try:
+        text = raw.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise SpecError(f"OpenAPI document at {source} is not valid UTF-8") from exc
+
+    document: Any = None
+    try:
+        document = json.loads(text)
+    except json.JSONDecodeError:
+        try:
+            document = yaml.safe_load(text)
+        except yaml.YAMLError as exc:
+            raise SpecError(
+                f"OpenAPI document at {source} is neither valid JSON nor valid YAML: {exc}"
+            ) from exc
+
+    if not isinstance(document, dict):
+        raise SpecError(
+            f"OpenAPI document at {source} did not parse to an object "
+            f"(got {type(document).__name__})"
+        )
+    return document