feat: add MCP (Model Context Protocol) support

Add comprehensive MCP integration using langchain-mcp-adapters to enable
pluggable external tools from MCP servers.

Features:
- MCP server configuration via mcp_config.json
- Automatic lazy initialization for seamless use in both FastAPI and LangGraph Studio
- Support for multiple MCP servers (filesystem, postgres, github, brave-search, etc.)
- Environment variable resolution in configuration
- Tool caching mechanism for optimal performance
- Complete documentation and setup guide

Implementation:
- Add src/mcp module with client, tools, and cache components
- Integrate MCP config loading in AppConfig
- Update tool system to include MCP tools automatically
- Add eager initialization in FastAPI lifespan handler
- Add lazy initialization fallback for LangGraph Studio

Dependencies:
- Add langchain-mcp-adapters>=0.1.0

Documentation:
- Add MCP_SETUP.md with comprehensive setup guide
- Update CLAUDE.md with MCP system architecture
- Update config.example.yaml with MCP configuration notes
- Update README.md with MCP setup instructions

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
hetaoBackend
2026-01-19 18:57:16 +08:00
parent 21f35b1d3c
commit 74d4a16492
19 changed files with 1044 additions and 5 deletions

View File

@@ -1,4 +1,5 @@
from .app_config import get_app_config
from .mcp_config import McpConfig, get_mcp_config
from .skills_config import SkillsConfig
__all__ = ["get_app_config", "SkillsConfig"]
__all__ = ["get_app_config", "SkillsConfig", "McpConfig", "get_mcp_config"]

View File

@@ -6,6 +6,7 @@ import yaml
from dotenv import load_dotenv
from pydantic import BaseModel, ConfigDict, Field
from src.config.mcp_config import McpConfig
from src.config.model_config import ModelConfig
from src.config.sandbox_config import SandboxConfig
from src.config.skills_config import SkillsConfig
@@ -24,6 +25,7 @@ class AppConfig(BaseModel):
tools: list[ToolConfig] = Field(default_factory=list, description="Available tools")
tool_groups: list[ToolGroupConfig] = Field(default_factory=list, description="Available tool groups")
skills: SkillsConfig = Field(default_factory=SkillsConfig, description="Skills configuration")
mcp: McpConfig = Field(default_factory=McpConfig, description="MCP configuration")
model_config = ConfigDict(extra="allow", frozen=False)
@classmethod
@@ -80,6 +82,10 @@ class AppConfig(BaseModel):
if "summarization" in config_data:
load_summarization_config_from_dict(config_data["summarization"])
# Load MCP config separately (it's in a different file)
mcp_config = McpConfig.from_file()
config_data["mcp"] = mcp_config.model_dump()
result = cls.model_validate(config_data)
return result

View File

@@ -0,0 +1,186 @@
"""MCP (Model Context Protocol) configuration."""
import json
import os
from pathlib import Path
from typing import Any
from pydantic import BaseModel, ConfigDict, Field
class McpServerConfig(BaseModel):
"""Configuration for a single MCP server."""
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")
model_config = ConfigDict(extra="allow")
class McpConfig(BaseModel):
"""Configuration for all MCP servers."""
mcp_servers: dict[str, McpServerConfig] = Field(
default_factory=dict,
description="Map of MCP server name to configuration",
alias="mcpServers",
)
model_config = ConfigDict(extra="allow", populate_by_name=True)
@classmethod
def resolve_config_path(cls, config_path: str | None = None) -> Path | None:
"""Resolve the MCP config file path.
Priority:
1. If provided `config_path` argument, use it.
2. If provided `DEER_FLOW_MCP_CONFIG_PATH` environment variable, use it.
3. Otherwise, check for `mcp_config.json` in the current directory, then in the parent directory.
4. If not found, return None (MCP is optional).
Args:
config_path: Optional path to MCP config file.
Returns:
Path to the MCP config file if found, otherwise None.
"""
if config_path:
path = Path(config_path)
if not path.exists():
raise FileNotFoundError(f"MCP config file specified by param `config_path` not found at {path}")
return path
elif os.getenv("DEER_FLOW_MCP_CONFIG_PATH"):
path = Path(os.getenv("DEER_FLOW_MCP_CONFIG_PATH"))
if not path.exists():
raise FileNotFoundError(f"MCP config file specified by environment variable `DEER_FLOW_MCP_CONFIG_PATH` not found at {path}")
return path
else:
# Check if the mcp_config.json is in the current directory
path = Path(os.getcwd()) / "mcp_config.json"
if path.exists():
return path
# Check if the mcp_config.json is in the parent directory of CWD
path = Path(os.getcwd()).parent / "mcp_config.json"
if path.exists():
return path
# MCP is optional, so return None if not found
return None
@classmethod
def from_file(cls, config_path: str | None = None) -> "McpConfig":
"""Load MCP config from JSON file.
See `resolve_config_path` for more details.
Args:
config_path: Path to the MCP config file.
Returns:
McpConfig: The loaded config, or empty config if file not found.
"""
resolved_path = cls.resolve_config_path(config_path)
if resolved_path is None:
# Return empty config if MCP config file is not found
return cls(mcp_servers={})
with open(resolved_path) as f:
config_data = json.load(f)
cls.resolve_env_variables(config_data)
return cls.model_validate(config_data)
@classmethod
def resolve_env_variables(cls, config: dict[str, Any]) -> dict[str, Any]:
"""Recursively resolve environment variables in the config.
Environment variables are resolved using the `os.getenv` function. Example: $OPENAI_API_KEY
Args:
config: The config to resolve environment variables in.
Returns:
The config with environment variables resolved.
"""
for key, value in config.items():
if isinstance(value, str):
if value.startswith("$"):
env_value = os.getenv(value[1:], None)
if env_value is not None:
config[key] = env_value
else:
config[key] = value
elif isinstance(value, dict):
config[key] = cls.resolve_env_variables(value)
elif isinstance(value, list):
config[key] = [cls.resolve_env_variables(item) if isinstance(item, dict) else item for item in value]
return config
def get_enabled_servers(self) -> dict[str, McpServerConfig]:
"""Get only the enabled MCP servers.
Returns:
Dictionary of enabled MCP servers.
"""
return {name: config for name, config in self.mcp_servers.items() if config.enabled}
_mcp_config: McpConfig | None = None
def get_mcp_config() -> McpConfig:
"""Get the MCP config instance.
Returns a cached singleton instance. Use `reload_mcp_config()` to reload
from file, or `reset_mcp_config()` to clear the cache.
Returns:
The cached McpConfig instance.
"""
global _mcp_config
if _mcp_config is None:
_mcp_config = McpConfig.from_file()
return _mcp_config
def reload_mcp_config(config_path: str | None = None) -> McpConfig:
"""Reload the MCP config from file and update the cached instance.
This is useful when the config file has been modified and you want
to pick up the changes without restarting the application.
Args:
config_path: Optional path to MCP config file. If not provided,
uses the default resolution strategy.
Returns:
The newly loaded McpConfig instance.
"""
global _mcp_config
_mcp_config = McpConfig.from_file(config_path)
return _mcp_config
def reset_mcp_config() -> None:
"""Reset the cached MCP config instance.
This clears the singleton cache, causing the next call to
`get_mcp_config()` to reload from file. Useful for testing
or when switching between different configurations.
"""
global _mcp_config
_mcp_config = None
def set_mcp_config(config: McpConfig) -> None:
"""Set a custom MCP config instance.
This allows injecting a custom or mock config for testing purposes.
Args:
config: The McpConfig instance to use.
"""
global _mcp_config
_mcp_config = config