214 lines
9.4 KiB
Python
214 lines
9.4 KiB
Python
import logging
|
|
from pathlib import Path
|
|
from typing import Literal
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
from app.gateway.deps import get_optional_user_from_request
|
|
from deerflow.config.extensions_config import (
|
|
ExtensionsConfig,
|
|
patch_extensions_config,
|
|
read_extensions_config_document,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api", tags=["mcp"])
|
|
|
|
|
|
async def _require_admin(request: Request) -> None:
|
|
"""Reject the request unless the caller is an authenticated admin user.
|
|
|
|
Allows the call through silently when auth is disabled (no user attached),
|
|
so existing single-user / no-auth deployments keep working.
|
|
"""
|
|
user = await get_optional_user_from_request(request)
|
|
if user is None:
|
|
# No authenticated user — auth might be disabled or middleware bypassed.
|
|
# Defer to the AuthMiddleware decision and let the call through.
|
|
return
|
|
if getattr(user, "system_role", None) != "admin":
|
|
raise HTTPException(status_code=403, detail="MCP configuration requires admin privileges")
|
|
|
|
|
|
class McpOAuthConfigResponse(BaseModel):
|
|
"""OAuth configuration for an MCP server."""
|
|
|
|
enabled: bool = Field(default=True, description="Whether OAuth token injection is enabled")
|
|
token_url: str = Field(default="", description="OAuth token endpoint URL")
|
|
grant_type: Literal["client_credentials", "refresh_token"] = Field(default="client_credentials", description="OAuth grant type")
|
|
client_id: str | None = Field(default=None, description="OAuth client ID")
|
|
client_secret: str | None = Field(default=None, description="OAuth client secret")
|
|
refresh_token: str | None = Field(default=None, description="OAuth refresh token")
|
|
scope: str | None = Field(default=None, description="OAuth scope")
|
|
audience: str | None = Field(default=None, description="OAuth audience")
|
|
token_field: str = Field(default="access_token", description="Token response field containing access token")
|
|
token_type_field: str = Field(default="token_type", description="Token response field containing token type")
|
|
expires_in_field: str = Field(default="expires_in", description="Token response field containing expires-in seconds")
|
|
default_token_type: str = Field(default="Bearer", description="Default token type when response omits token_type")
|
|
refresh_skew_seconds: int = Field(default=60, description="Refresh this many seconds before expiry")
|
|
extra_token_params: dict[str, str] = Field(default_factory=dict, description="Additional form params sent to token endpoint")
|
|
model_config = ConfigDict(extra="allow", populate_by_name=True)
|
|
|
|
|
|
class McpServerConfigResponse(BaseModel):
|
|
"""Response model for MCP server configuration."""
|
|
|
|
enabled: bool = Field(default=True, description="Whether this MCP server is enabled")
|
|
type: str = Field(default="stdio", description="Transport type: 'stdio', 'sse', or 'http'")
|
|
command: str | None = Field(default=None, description="Command to execute to start the MCP server (for stdio type)")
|
|
args: list[str] = Field(default_factory=list, description="Arguments to pass to the command (for stdio type)")
|
|
env: dict[str, str] = Field(default_factory=dict, description="Environment variables for the MCP server")
|
|
url: str | None = Field(default=None, description="URL of the MCP server (for sse or http type)")
|
|
headers: dict[str, str] = Field(default_factory=dict, description="HTTP headers to send (for sse or http type)")
|
|
oauth: McpOAuthConfigResponse | None = Field(default=None, description="OAuth configuration for MCP HTTP/SSE servers")
|
|
description: str = Field(default="", description="Human-readable description of what this MCP server provides")
|
|
disabled_tools: list[str] = Field(
|
|
default_factory=list,
|
|
description="Tool names (without server-name prefix) to hide from the agent",
|
|
)
|
|
model_config = ConfigDict(extra="allow", populate_by_name=True)
|
|
|
|
|
|
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(http_request: Request) -> 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"
|
|
}
|
|
}
|
|
}
|
|
```
|
|
"""
|
|
await _require_admin(http_request)
|
|
config_path = ExtensionsConfig.resolve_config_path()
|
|
if config_path is None:
|
|
return McpConfigResponse()
|
|
document = read_extensions_config_document(config_path)
|
|
raw_servers = document.get("mcpServers", {})
|
|
if not isinstance(raw_servers, dict):
|
|
raise HTTPException(status_code=500, detail="Extensions config 'mcpServers' must be a JSON object")
|
|
return McpConfigResponse(
|
|
mcp_servers={name: McpServerConfigResponse.model_validate(server) for name, server in raw_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, http_request: Request) -> 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"
|
|
}
|
|
}
|
|
}
|
|
```
|
|
"""
|
|
await _require_admin(http_request)
|
|
|
|
# Light validation: each server must have the fields required by its transport.
|
|
for server_name, server in request.mcp_servers.items():
|
|
if not server_name.strip():
|
|
raise HTTPException(status_code=400, detail="MCP server name cannot be empty")
|
|
transport = (server.type or "stdio").lower()
|
|
if transport == "stdio":
|
|
if not (server.command or "").strip():
|
|
raise HTTPException(status_code=400, detail=f"MCP server '{server_name}' uses stdio transport but command is empty")
|
|
elif transport in ("http", "sse"):
|
|
if not (server.url or "").strip():
|
|
raise HTTPException(status_code=400, detail=f"MCP server '{server_name}' uses {transport} transport but url is empty")
|
|
else:
|
|
raise HTTPException(status_code=400, detail=f"MCP server '{server_name}' has unsupported transport: {transport}")
|
|
|
|
try:
|
|
# Get the current config path (or determine where to save it)
|
|
config_path = ExtensionsConfig.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 / "extensions_config.json"
|
|
logger.info(f"No existing extensions config found. Creating new config at: {config_path}")
|
|
|
|
mcp_servers = {name: server.model_dump(by_alias=True) for name, server in request.mcp_servers.items()}
|
|
|
|
def _patch_mcp_servers(config_data: dict) -> None:
|
|
config_data["mcpServers"] = mcp_servers
|
|
|
|
patch_extensions_config(config_path, _patch_mcp_servers)
|
|
|
|
logger.info(f"MCP configuration updated and saved to: {config_path}")
|
|
|
|
# The shared writer reloads this process's cache. LangGraph Server runs
|
|
# separately and detects the file change via mtime.
|
|
|
|
# Echo the validated raw request rather than the reloaded model, whose
|
|
# env placeholders are expanded in memory and must never be persisted
|
|
# by a later GET -> PUT round trip.
|
|
return McpConfigResponse(mcp_servers=request.mcp_servers)
|
|
|
|
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)}")
|