When loading multiple MCP Servers from config file, add links jumping to tool's docs from root docs, also add link to jump back.

for example:

http://127.0.0.1:8000/docs

```markdown
Automatically generated API from MCP Tool Schemas

available tools:
- memory
- time
```

http://127.0.0.1:8000/memory/docs

```markdown
memory MCP Server

- back to tool list
```
This commit is contained in:
旺 杜 2025-04-04 15:13:54 +08:00
parent f1b5fbd52c
commit 9854e48205

View File

@ -1,18 +1,16 @@
from fastapi import FastAPI, Body, Depends
from fastapi.middleware.cors import CORSMiddleware
from starlette.routing import Mount
from pydantic import create_model
from contextlib import AsyncExitStack, asynccontextmanager
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client
from typing import Dict, Any, Optional
import uvicorn
import json
import os
from contextlib import AsyncExitStack, asynccontextmanager
from typing import Dict, Any, Optional
import uvicorn
from fastapi import FastAPI, Body, Depends
from fastapi.middleware.cors import CORSMiddleware
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client
from mcpo.utils.auth import get_verify_api_key
from pydantic import create_model
from starlette.routing import Mount
def get_python_type(param_type: str):
@ -42,9 +40,7 @@ async def create_dynamic_endpoints(app: FastAPI, api_dependency=None):
server_info = getattr(result, "serverInfo", None)
if server_info:
app.title = server_info.name or app.title
app.description = (
f"{server_info.name} MCP Server" if server_info.name else app.description
)
app.description = app.description
app.version = server_info.version or app.version
tools_result = await session.list_tools()
@ -139,13 +135,12 @@ async def lifespan(app: FastAPI):
async def run(
host: str = "127.0.0.1",
port: int = 8000,
api_key: Optional[str] = "",
cors_allow_origins=["*"],
**kwargs,
host: str = "127.0.0.1",
port: int = 8000,
api_key: Optional[str] = "",
cors_allow_origins=["*"],
**kwargs,
):
# Server API Key
api_dependency = get_verify_api_key(api_key) if api_key else None
@ -154,11 +149,11 @@ async def run(
server_command = kwargs.get("server_command")
name = kwargs.get("name") or "MCP OpenAPI Proxy"
description = (
kwargs.get("description") or "Automatically generated API from MCP Tool Schemas"
kwargs.get("description") or "Automatically generated API from MCP Tool Schemas"
)
version = kwargs.get("version") or "1.0"
ssl_certfile = kwargs.get("ssl_certfile")
ssl_keyfile= kwargs.get("ssl_keyfile")
ssl_keyfile = kwargs.get("ssl_keyfile")
main_app = FastAPI(
title=name, description=description, version=version, ssl_certfile=ssl_certfile, ssl_keyfile=ssl_keyfile, lifespan=lifespan
@ -183,14 +178,13 @@ async def run(
with open(config_path, "r") as f:
config_data = json.load(f)
mcp_servers = config_data.get("mcpServers", {})
if not mcp_servers:
raise ValueError("No 'mcpServers' found in config file.")
main_app.description += "\n\n- **available tools**"
for server_name, server_cfg in mcp_servers.items():
sub_app = FastAPI(
title=f"{server_name}",
description=f"{server_name} MCP Server",
description=f"{server_name} MCP Server\n\n- [back to tool list](http://{host}:{port}/docs)",
version="1.0",
lifespan=lifespan,
)
@ -208,13 +202,12 @@ async def run(
sub_app.state.env = {**os.environ, **server_cfg.get("env", {})}
sub_app.state.api_dependency = api_dependency
main_app.mount(f"/{server_name}", sub_app)
main_app.description += f"\n - [{server_name}](http://{host}:{port}/{server_name}/docs)"
else:
raise ValueError("You must provide either server_command or config.")
config = uvicorn.Config(app=main_app, host=host, port=port, ssl_certfile=ssl_certfile , ssl_keyfile=ssl_keyfile ,log_level="info")
config = uvicorn.Config(app=main_app, host=host, port=port, ssl_certfile=ssl_certfile, ssl_keyfile=ssl_keyfile, log_level="info")
server = uvicorn.Server(config)
await server.serve()