feat: add optional tools allowlist via MCP_TOOLS_INCLUDED
When MCP_TOOLS_INCLUDED is set to a comma-separated list of tool names, only those tools are exposed (tools/list and calls); unset or empty exposes every generated tool. Unknown names are ignored with a startup warning and the active selection is logged at startup.
This commit is contained in:
@@ -33,6 +33,10 @@ MCP_HOST=0.0.0.0
|
||||
MCP_PORT=8000
|
||||
#MCP_SERVER_NAME=My API
|
||||
|
||||
# Comma-separated allowlist of tool names to expose. Unset or empty exposes
|
||||
# every generated tool; unknown names are ignored with a startup warning.
|
||||
#MCP_TOOLS_INCLUDED=getPetById,findPetsByStatus
|
||||
|
||||
# Timeout in seconds for fetching the spec and calling the API (default: 30).
|
||||
#HTTP_TIMEOUT=30
|
||||
|
||||
|
||||
@@ -82,6 +82,7 @@ OPENAPI_SPEC_URL=https://petstore3.swagger.io/api/v3/openapi.json uv run xmcp
|
||||
| `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 |
|
||||
| `HTTP_TIMEOUT` | no | `30` | Timeout in seconds for spec fetch and API calls |
|
||||
| `LOG_LEVEL` | no | `INFO` | Python log level |
|
||||
|
||||
@@ -116,6 +117,20 @@ 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 allowlist
|
||||
|
||||
By default every operation in the OpenAPI document is exposed as an MCP tool.
|
||||
Set `MCP_TOOLS_INCLUDED` to a comma-separated list of tool names to expose
|
||||
only those:
|
||||
|
||||
```sh
|
||||
MCP_TOOLS_INCLUDED=getPetById,findPetsByStatus
|
||||
```
|
||||
|
||||
Unset or empty exposes all tools. Names that do not match a generated tool are
|
||||
ignored with a startup warning, and the active selection is reported at startup
|
||||
(`Tools allowlist active: exposing N of M tool(s): ...`).
|
||||
|
||||
### Serving a local spec file
|
||||
|
||||
```sh
|
||||
|
||||
@@ -62,6 +62,7 @@ class Config:
|
||||
api_auth_header: str
|
||||
api_auth_scheme: str
|
||||
mcp_api_key_header: str | None
|
||||
tools_included: frozenset[str] | None
|
||||
transport: str
|
||||
host: str
|
||||
port: int
|
||||
@@ -101,6 +102,15 @@ class Config:
|
||||
key_header = "X-API-Key" if raw_key_header is None else raw_key_header.strip()
|
||||
key_header = key_header or None
|
||||
|
||||
# Comma-separated allowlist of tool names; unset or empty exposes all.
|
||||
raw_included = _get("MCP_TOOLS_INCLUDED")
|
||||
included: frozenset[str] | None = None
|
||||
if raw_included is not None:
|
||||
names = frozenset(
|
||||
name.strip() for name in raw_included.split(",") if name.strip()
|
||||
)
|
||||
included = names or None
|
||||
|
||||
return cls(
|
||||
spec_url=spec_url,
|
||||
spec_token=_get("OPENAPI_SPEC_TOKEN"),
|
||||
@@ -109,6 +119,7 @@ class Config:
|
||||
api_auth_header=_get("API_AUTH_HEADER") or "Authorization",
|
||||
api_auth_scheme=auth_scheme,
|
||||
mcp_api_key_header=key_header,
|
||||
tools_included=included,
|
||||
transport=transport,
|
||||
host=_get("MCP_HOST") or "0.0.0.0",
|
||||
port=_get_int("MCP_PORT", 8000, minimum=1, maximum=65535),
|
||||
|
||||
+32
-12
@@ -54,17 +54,37 @@ def build_server(config: Config) -> FastMCP:
|
||||
return server
|
||||
|
||||
|
||||
async def _log_available_tools(server: FastMCP) -> None:
|
||||
"""Log the tools generated from the OpenAPI document (DEBUG level)."""
|
||||
async def _prepare_tools(server: FastMCP, included: frozenset[str] | None) -> None:
|
||||
"""Apply the optional tools allowlist and log the exposed tools."""
|
||||
tools = await server.list_tools()
|
||||
unique: dict[str, object] = {}
|
||||
for tool in tools:
|
||||
unique.setdefault(tool.name, tool)
|
||||
logger.debug("Available MCP tools (%d):", len(unique))
|
||||
for tool in unique.values():
|
||||
description = (tool.description or "").strip().splitlines()
|
||||
suffix = f" — {description[0]}" if description else ""
|
||||
logger.debug(" - %s%s", tool.name, suffix)
|
||||
|
||||
if included is not None:
|
||||
known = {tool.name for tool in tools}
|
||||
unknown = sorted(included - known)
|
||||
if unknown:
|
||||
logger.warning(
|
||||
"MCP_TOOLS_INCLUDED lists unknown tool(s): %s",
|
||||
", ".join(unknown),
|
||||
)
|
||||
exposed = included & known
|
||||
server.enable(names=set(exposed), only=True, components={"tool"})
|
||||
tools = await server.list_tools()
|
||||
logger.info(
|
||||
"Tools allowlist active: exposing %d of %d tool(s)%s",
|
||||
len(exposed),
|
||||
len(known),
|
||||
f": {', '.join(sorted(exposed))}" if exposed else "",
|
||||
)
|
||||
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
unique: dict[str, object] = {}
|
||||
for tool in tools:
|
||||
unique.setdefault(tool.name, tool)
|
||||
logger.debug("Available MCP tools (%d):", len(unique))
|
||||
for tool in unique.values():
|
||||
description = (tool.description or "").strip().splitlines()
|
||||
suffix = f" — {description[0]}" if description else ""
|
||||
logger.debug(" - %s%s", tool.name, suffix)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
@@ -85,8 +105,8 @@ def main() -> None:
|
||||
except SpecError as exc:
|
||||
raise SystemExit(f"Failed to load the OpenAPI document: {exc}") from exc
|
||||
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
asyncio.run(_log_available_tools(server))
|
||||
if config.tools_included is not None or logger.isEnabledFor(logging.DEBUG):
|
||||
asyncio.run(_prepare_tools(server, config.tools_included))
|
||||
|
||||
# Logs go to stderr so the stdio transport's stdout stays protocol-clean.
|
||||
if config.transport == "http":
|
||||
|
||||
Reference in New Issue
Block a user