feat: add MCP API endpoint and enhance API documentation

Add new MCP configuration management endpoint and enhance API documentation
with detailed descriptions, examples, and OpenAPI support for better
developer experience.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
hetaoBackend
2026-01-20 13:20:50 +08:00
parent d11763dcc8
commit 8434cf4c60
6 changed files with 307 additions and 11 deletions

View File

@@ -0,0 +1,148 @@
import json
import logging
from pathlib import Path
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
from src.config.mcp_config import McpConfig, get_mcp_config, reload_mcp_config
from src.mcp.cache import reset_mcp_tools_cache
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api", tags=["mcp"])
class McpServerConfigResponse(BaseModel):
"""Response model for MCP server configuration."""
enabled: bool = Field(default=True, description="Whether this MCP server is enabled")
command: str = Field(..., description="Command to execute to start the MCP server")
args: list[str] = Field(default_factory=list, description="Arguments to pass to the command")
env: dict[str, str] = Field(default_factory=dict, description="Environment variables for the MCP server")
description: str = Field(default="", description="Human-readable description of what this MCP server provides")
class McpConfigResponse(BaseModel):
"""Response model for MCP configuration."""
mcp_servers: dict[str, McpServerConfigResponse] = Field(
default_factory=dict,
description="Map of MCP server name to configuration",
)
class McpConfigUpdateRequest(BaseModel):
"""Request model for updating MCP configuration."""
mcp_servers: dict[str, McpServerConfigResponse] = Field(
...,
description="Map of MCP server name to configuration",
)
@router.get(
"/mcp/config",
response_model=McpConfigResponse,
summary="Get MCP Configuration",
description="Retrieve the current Model Context Protocol (MCP) server configurations.",
)
async def get_mcp_configuration() -> McpConfigResponse:
"""Get the current MCP configuration.
Returns:
The current MCP configuration with all servers.
Example:
```json
{
"mcp_servers": {
"github": {
"enabled": true,
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-github"],
"env": {"GITHUB_TOKEN": "ghp_xxx"},
"description": "GitHub MCP server for repository operations"
}
}
}
```
"""
config = get_mcp_config()
return McpConfigResponse(
mcp_servers={name: McpServerConfigResponse(**server.model_dump()) for name, server in config.mcp_servers.items()}
)
@router.put(
"/mcp/config",
response_model=McpConfigResponse,
summary="Update MCP Configuration",
description="Update Model Context Protocol (MCP) server configurations and save to file.",
)
async def update_mcp_configuration(request: McpConfigUpdateRequest) -> McpConfigResponse:
"""Update the MCP configuration.
This will:
1. Save the new configuration to the mcp_config.json file
2. Reload the configuration cache
3. Reset MCP tools cache to trigger reinitialization
Args:
request: The new MCP configuration to save.
Returns:
The updated MCP configuration.
Raises:
HTTPException: 500 if the configuration file cannot be written.
Example Request:
```json
{
"mcp_servers": {
"github": {
"enabled": true,
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-github"],
"env": {"GITHUB_TOKEN": "$GITHUB_TOKEN"},
"description": "GitHub MCP server for repository operations"
}
}
}
```
"""
try:
# Get the current config path (or determine where to save it)
config_path = McpConfig.resolve_config_path()
# If no config file exists, create one in the parent directory (project root)
if config_path is None:
config_path = Path.cwd().parent / "mcp_config.json"
logger.info(f"No existing MCP config found. Creating new config at: {config_path}")
# Convert request to dict format for JSON serialization
config_data = {"mcpServers": {name: server.model_dump() for name, server in request.mcp_servers.items()}}
# Write the configuration to file
with open(config_path, "w") as f:
json.dump(config_data, f, indent=2)
logger.info(f"MCP configuration updated and saved to: {config_path}")
# Reload the configuration to update the cache
reload_mcp_config()
# Reset MCP tools cache so they will be reinitialized with new config on next use
reset_mcp_tools_cache()
logger.info("MCP tools cache reset - tools will be reinitialized on next use")
# Return the updated configuration
reloaded_config = get_mcp_config()
return McpConfigResponse(
mcp_servers={name: McpServerConfigResponse(**server.model_dump()) for name, server in reloaded_config.mcp_servers.items()}
)
except Exception as e:
logger.error(f"Failed to update MCP configuration: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to update MCP configuration: {str(e)}")