feat: accept per-client API keys from an HTTP header

When serving over HTTP, the key sent in the MCP_API_KEY_HEADER header
(default X-API-Key) is used as the API credential for that request and
takes precedence over the static API_TOKEN. The key is formatted with
API_AUTH_HEADER/API_AUTH_SCHEME and the source header is stripped from
the outgoing API request. Implemented as an httpx2 auth flow over
FastMCP's request context, so concurrent clients stay isolated.

.env.example

@@ -21,6 +21,11 @@ OPENAPI_SPEC_URL=https://petstore3.swagger.io/api/v3/openapi.json
 # 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

README.md

@@ -77,6 +77,7 @@ OPENAPI_SPEC_URL=https://petstore3.swagger.io/api/v3/openapi.json uv run xmcp
 | `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_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 |
@@ -88,6 +89,32 @@ 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.
 
+### Per-client API keys (HTTP transport)
+
+When serving over HTTP, clients can send their own API key on each request
+instead of sharing the server's `API_TOKEN`. The value is treated as a raw
+token and forwarded to the target API using `API_AUTH_HEADER` /
+`API_AUTH_SCHEME` (by default `Authorization: Bearer <key>`):
+
+```sh
+curl -X POST http://localhost:8000/mcp \
+  -H 'Content-Type: application/json' \
+  -H 'Accept: application/json, text/event-stream' \
+  -H 'X-API-Key: sk-client-secret' \
+  -d '{"jsonrpc":"2.0","id":1,"method":"initialize",...}'
+```
+
+Precedence: the request header wins, then `API_TOKEN`, then no authentication.
+The header name is configurable with `MCP_API_KEY_HEADER`; set it to an empty
+value to disable the mapping. When the key is mapped to a differently named
+target header, the source header is removed from the outgoing API request so
+the credential is not sent twice.
+
+Note: FastMCP independently passes other incoming HTTP headers through to the
+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`).
+
 ### Serving a local spec file
 
 ```sh
@@ -117,6 +144,9 @@ The base URL for generated tools is chosen as follows:
 - 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.
+- Per-request API keys (the `X-API-Key` header by default) are forwarded only to
+  the target API and are never logged. Terminate TLS in front of the MCP
+  endpoint so client keys are not sent in clear text.
 
 ## Releasing
 
@@ -136,6 +166,7 @@ This publishes `gitea.woggioni.net/woggioni/xmcp:0.1.0` and
 ```
 src/xmcp/config.py        env var parsing and validation
 src/xmcp/spec.py          spec loading, base URL resolution, API client
+src/xmcp/auth.py          per-request API key forwarding (HTTP transport)
 src/xmcp/server.py        FastMCP server construction and entrypoint
 Dockerfile                multi-stage build (uv builder -> alpine runtime, non-root)
 .gitea/workflows/build.yaml  release pipeline (tag push -> docker build & push)

src/xmcp/auth.py

@@ -0,0 +1,54 @@
+"""Per-request API credentials forwarded from incoming MCP HTTP requests."""
+
+from __future__ import annotations
+
+from collections.abc import Generator
+
+import httpx2
+from fastmcp.server.dependencies import get_http_headers
+
+
+def format_credential(token: str, scheme: str) -> str:
+    """Format a raw token with the configured authentication scheme."""
+    return f"{scheme} {token}".strip() if scheme else token
+
+
+class ForwardedTokenAuth(httpx2.Auth):
+    """Use the API key supplied by the MCP client for the current request.
+
+    The token is read from ``source_header`` on the HTTP request that triggered
+    the tool call and written to ``target_header`` on the outgoing API request.
+    When the incoming header is absent, any statically configured header on the
+    client is left untouched, so the environment token remains the fallback.
+
+    Note: FastMCP already forwards non-credential custom headers from the MCP
+    request to the API request. When this auth flow maps a credential to a
+    different header name, the original header is removed from the outgoing
+    request so the key is not sent twice.
+    """
+
+    def __init__(self, *, source_header: str, target_header: str, scheme: str) -> None:
+        self._source_header = source_header.lower()
+        self._target_header = target_header
+        self._scheme = scheme
+
+    def _forwarded_value(self) -> str | None:
+        # `include` overrides FastMCP's default stripping of credential headers,
+        # which matters when the configured header is `Authorization`.
+        headers = get_http_headers(include={self._source_header})
+        for name, value in headers.items():
+            if name.lower() != self._source_header:
+                continue
+            value = value.strip()
+            return format_credential(value, self._scheme) if value else None
+        return None
+
+    def auth_flow(
+        self, request: httpx2.Request
+    ) -> Generator[httpx2.Request, httpx2.Response, None]:
+        value = self._forwarded_value()
+        if value is not None:
+            request.headers[self._target_header] = value
+            if self._source_header != self._target_header.lower():
+                request.headers.pop(self._source_header, None)
+        yield request

src/xmcp/config.py

@@ -61,6 +61,7 @@ class Config:
     api_token: str | None
     api_auth_header: str
     api_auth_scheme: str
+    mcp_api_key_header: str | None
     transport: str
     host: str
     port: int
@@ -95,6 +96,11 @@ class Config:
         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
+
         return cls(
             spec_url=spec_url,
             spec_token=_get("OPENAPI_SPEC_TOKEN"),
@@ -102,6 +108,7 @@ class Config:
             api_token=_get("API_TOKEN"),
             api_auth_header=_get("API_AUTH_HEADER") or "Authorization",
             api_auth_scheme=auth_scheme,
+            mcp_api_key_header=key_header,
             transport=transport,
             host=_get("MCP_HOST") or "0.0.0.0",
             port=_get_int("MCP_PORT", 8000, minimum=1, maximum=65535),

src/xmcp/server.py

@@ -23,6 +23,7 @@ def build_server(config: Config) -> FastMCP:
         header=config.api_auth_header,
         scheme=config.api_auth_scheme,
         timeout=config.timeout,
+        forwarded_header=config.mcp_api_key_header,
     )
 
     info = spec.get("info") if isinstance(spec.get("info"), dict) else {}
@@ -38,11 +39,16 @@ def build_server(config: Config) -> FastMCP:
     )
     if config.api_token:
         logger.debug(
-            "API requests carry the %s header (%s)",
+            "Fallback API token configured for the %s header (%s)",
             config.api_auth_header,
             config.api_auth_scheme or "raw token",
         )
-    else:
+    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
 

src/xmcp/spec.py

@@ -13,6 +13,8 @@ from urllib.request import url2pathname
 import httpx2
 import yaml
 
+from .auth import ForwardedTokenAuth, format_credential
+
 logger = logging.getLogger("xmcp.spec")
 
 _REMOTE_SCHEMES = ("http", "https")
@@ -78,12 +80,18 @@ def build_api_client(
     header: str,
     scheme: str,
     timeout: float,
+    forwarded_header: str | None,
 ) -> 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)
+        headers[header] = format_credential(token, scheme)
+    auth = (
+        ForwardedTokenAuth(source_header=forwarded_header, target_header=header, scheme=scheme)
+        if forwarded_header
+        else None
+    )
+    return httpx2.AsyncClient(base_url=base_url, headers=headers, timeout=timeout, auth=auth)
 
 
 def _fetch_remote(url: str, *, token: str | None, timeout: float) -> bytes: