MCP_understanding

MCP_understanding

MCP 服务与 A2A 工具 — 全面深入探究

目录

1. 执行摘要
2. 项目架构概览
3. 模型上下文协议（MCP）——详解
4. 核心组件与实现
5. 技术栈
6. MCP 服务实现
7. A2A（Agent-to-Agent）集成
8. 高级实现
9. 部署与配置
10. 用例与示例
11. 最佳实践与经验总结

---

执行摘要

本项目是一个 模型上下文协议（MCP）的全面探索与实现套件，并结合了 Google 的 Agent-to-Agent（A2A）框架。它展示了如何构建、编排和部署配备多种专用工具与服务的 AI Agent。

主要亮点：

• 20 多种 MCP 服务实现（FastMCP、各种特定领域的服务）
• 与 Google ADK（Agent Development Kit）全面集成
• 支持 A2A 协议，实现复杂的 Agent 编排
• 真实场景示例：货币转换、天气、YouTube 字幕提取、Neo4j GraphRAG
• SSE（服务器发送事件）和 Stdio 传输协议
• 多 Agent 协调模式
• 具备错误处理和可观测性的生产就绪模式

---

项目架构概览

高层架构

┌─────────────────────────────────────────────────────────────────┐
│                    Client Applications                          │
│           (Claude Desktop, Custom Clients, UIs)                │
└────────────────────────────────────────────────────────────────┐
                            │
                ┌───────────┴───────────┐
                │                       │
         ┌──────▼──────┐        ┌──────▼──────┐
         │ MCP Servers │        │ A2A Agents  │
         │   (SPIs)    │        │  (Agentic)  │
         └──────┬──────┘        └──────┬──────┘
                │                      │
         ┌──────▼──────────────────────▼──────┐
         │      Transport Layer               │
         │  (Stdio, SSE, JSON-RPC 2.0)       │
         └──────┬──────────────────────┬──────┘
                │                      │
    ┌───────────▼──────┐   ┌──────────▼──────────┐
    │  External APIs   │   │  Data Sources       │
    │  - Web Search    │   │  - Neo4j Graph DB  │
    │  - Exchange Rates│   │  - BigQuery        │
    │  - Crypto Prices │   │  - PostgreSQL      │
    │  - Weather       │   │  - MySQL           │
    │  - YouTube       │   │  - Snowflake       │
    └──────────────────┘   └────────────────────┘

项目结构

MCP_servers_and_a2a_utils/
├── Core Entry Points
│   ├── agent.py                          # Multi-server MCP agent orchestration
│   ├── call_tool.py                      # Tool invocation wrapper
│   ├── list_tools.py                     # Tool discovery & introspection
│   └── debug_tools.py                    # Debugging utilities
│
├── MCP Servers (20+ implementations)
│   ├── mcp_weather_server/               # FastMCP weather service
│   ├── mcp_tavily_search/                # Web search integration
│   ├── mcp_math_server/                  # Mathematical operations
│   ├── mcp_youtube_extraction_server/    # YouTube transcript extraction
│   ├── mcp_database_server/              # Database operations
│   ├── mysql_mcp_server/                 # MySQL-specific server
│   ├── mcp_server_aws/                   # AWS integration
│   ├── mcp_snowflake/                    # Snowflake data warehouse
│   ├── neo4j-mcp-workspace-template/     # Graph database (5 specialized servers)
│   └── [15+ other specialized servers]
│
├── A2A Integration Layer
│   ├── a2a_src/                          # A2A protocol implementation
│   ├── agno_agentic_mcp/                 # Agno + Agentic framework
│   ├── google_adk_agent_mcp/             # Google ADK agents
│   │   ├── currency_agent.py             # Live A2A agent example
│   │   └── test_client.py                # A2A client test harness
│   └── [Multiple agentic frameworks]
│
├── Framework & Adapter Layer
│   ├── mcp_adapaters_clients/            # MCP client adapters
│   ├── fastmcp_fastapi/                  # FastMCP + FastAPI bridge
│   ├── langchain_mcp_server_client/      # LangChain integration
│   └── firebase/storage/remote/          # Remote storage backends
│
├── Documentation & Configuration
│   ├── README.md                         # Project overview
│   ├── MCP_Readme.md                     # MCP-specific details
│   ├── MCP_A2A_SSE_Studio.md             # SSE & A2A protocols
│   ├── MCP_INspect_TEST.md               # MCP Inspector usage
│   ├── claude_config.json                # Claude Desktop config
│   └── copilot-instructions.md           # Copilot integration
│
└── Configuration & Environment
    ├── pyproject.toml                    # Python project definition
    ├── requirements.txt                  # Core dependencies
    └── [Virtual environment configs]

---

模型上下文协议（MCP）——详解

什么是 MCP？

模型上下文协议是一种标准化的开放协议，用于将 AI 模型（如 Claude）与外部工具、数据源和上下文提供方连接起来。它使模型能够：

• 访问工具以执行操作
• 读取资源（文件、文档、数据）
• 调用提示词以进行结构化交互
• 请求采样（实验性功能）

MCP 架构

┌─────────────────┐                    ┌──────────────────┐
│   MCP Client    │◄──JSON-RPC 2.0─────► MCP Server       │
│  (e.g., Claude) │   (Stdio/SSE)       (Tool Provider)   │
└─────────────────┘                    └──────────────────┘
        │
    Uses:
        │
    ├─ Tools: Execute tasks
    ├─ Resources: Access data
    ├─ Prompts: Structured requests
    └─ Sampling: LLM interactions

核心概念

1. 工具

由服务端提供、客户端可调用的函数。示例：

@mcp.tool()
def get_exchange_rate(from_currency: str, to_currency: str) -> str:
    """Get current exchange rate between two currencies."""
    # Implementation
    return rate_data

特征：

• 使用 Pydantic 模型进行强类型定义
• 包含描述和注释
• 支持异步操作
• 返回结构化数据

2. 资源

客户端可读取的数据源。示例：

@mcp.resource("papers://folders")
def get_available_folders() -> str:
    """List available paper folders."""
    return folder_listing

类型：

• 基于文本（文档、日志）
• 二进制（图像、PDF）
• 分层结构（文件夹结构）

3. 提示词

带参数的可复用提示词模板：

@mcp.prompt()
def generate_search_prompt(topic: str, num_papers: int = 5) -> str:
    """Generate a prompt for finding academic papers."""
    return f"Find {num_papers} papers about {topic}..."

4. 工具注释

帮助客户端了解工具行为的元数据：

annotations=ToolAnnotations(
    title="Chunk Lexical Graph",
    readOnlyHint=False,           # Can modify data
    destructiveHint=False,        # Won't delete critical data
    idempotentHint=False,         # May have side effects
    openWorldHint=False,          # Operates on closed knowledge
)

传输协议

Stdio 传输

• **用途：**Unix 管道 / stdin/stdout
• **适用场景：**桌面应用、claude-desktop-config.json
• **特征：**同步请求-响应、低延迟
• 配置：

  {
    "command": "python",
    "args": ["server.py"],
    "transport": "stdio"
  }
  

SSE 传输（服务器发送事件）

• **用途：**使用 EventStream 的 HTTP
• **适用场景：**Web 服务、微服务、云部署
• **特征：**异步、双向流式传输
• 示例：

  mcp = FastMCP("currency", port=7080)
  await mcp.run_sse_async(host="0.0.0.0", port=7080)
  

MCP 协议流程

Client                                  Server
  │                                       │
  ├─ Initialize                          │
  │  {"jsonrpc": "2.0", "id": 1,        │
  │   "method": "initialize", ...}──────► 
  │                                    ┌─┤
  │                          ◄────────┤─ {"result": {"serverInfo": ...}}
  │                                    └─┤
  │
  ├─ List Tools                          │
  │  {"jsonrpc": "2.0", "id": 2,        │
  │   "method": "tools/list"}───────────►
  │                           ┌──────────┤
  │                  ◄────────┤ {"tools": [...]}
  │                           └──────────┤
  │
  ├─ Call Tool                           │
  │  {"jsonrpc": "2.0", "id": 3,        │
  │   "method": "tools/call",           │
  │   "params": {...}}─────────────────►
  │                     ┌────────────────┤
  │             ◄───────┤ {"result": "..."}
  │                     └────────────────┤

工具结果类型

MCP 工具会返回结构化结果：

# Success Result
{
    "type": "text",
    "text": "The exchange rate is 1.3794 CAD per USD"
}

# Tool Error
{
    "type": "text",
    "text": "Error: Invalid currency code",
    "isError": true
}

# Multi-part Result (images, formatted content)
{
    "type": "image",
    "data": "base64-encoded-image-data",
    "mimeType": "image/png"
}

---

核心组件与实现

1. FastMCP 框架

FastMCP 是本项目用于构建 MCP 服务的主要框架。

优势：

• 最少的样板代码
• 自动注册工具
• 内置异步支持
• 内置 SSE 和 Stdio 传输
• 集成 Pydantic，确保类型安全

基本结构：

from mcp.server.fastmcp import FastMCP

# Create server instance
mcp = FastMCP("MyServer", port=8000)

# Register tools
@mcp.tool()
async def my_tool(param: str) -> str:
    return f"Result: {param}"

# Register resources
@mcp.resource("docs://{path}")
def get_docs(path: str) -> str:
    return read_file(path)

# Run server
if __name__ == "__main__":
    mcp.run(transport='sse')  # or 'stdio'

2. LangChain MCP 适配器

将 MCP 服务与 LangChain Agent 连接起来：

from langchain_mcp_adapters.client import MultiServerMCPClient

async with MultiServerMCPClient({
    "tavily": {
        "command": "python",
        "args": ["servers/tavily.py"],
        "transport": "stdio"
    }
}) as client:
    tools = client.get_tools()
    agent = create_react_agent(model, tools)
    response = await agent.ainvoke({"messages": [...]})

3. MCP Inspector

用于测试和调试 MCP 服务的浏览器工具：

npx @modelcontextprotocol/inspector

# Starts on http://127.0.0.1:6274
# Test tools, resources, and prompts in real-time

功能：

• 可视化工具/资源浏览器
• 实时调用工具
• 参数验证
• 响应检查
• 错误诊断

---

技术栈

核心依赖

FastMCP 2.14.1              # MCP server framework
mcp                         # Official MCP protocol library
httpx                       # Async HTTP client
pydantic                    # Type validation & schemas

Optional Integrations:
- langchain-core            # LangChain framework
- langchain-mcp-adapters    # MCP-LangChain bridge
- langchain-openai          # OpenAI integration
- langgraph                  # Agent orchestration
- google.adk                 # Google Agent Development Kit
- neo4j                      # Graph database driver
- pymupdf                    # PDF processing
- docling                    # Document parsing
- pandas                     # Data manipulation
- requests                   # HTTP library
- beautifulsoup4             # HTML parsing

Python 生态集成

Python 3.11+
├─ Async/Await (asyncio, aiohttp)
├─ Type Hints (typing, pydantic)
├─ JSON-RPC 2.0 Protocol
├─ Environment Management (.zprofile, os.environ)
└─ Package Management (pip, uv)

---

MCP 服务实现

1. 货币与汇率服务

位置： google_adk_agent_mcp/

# Server: Provides exchange rate and crypto price tools
@mcp.tool()
def get_exchange_rate(currency_from: str, currency_to: str) -> dict:
    """Get real-time exchange rates via ExchangeRate-API."""
    
@mcp.tool()
def get_crypto_prices(symbols: str, vs_currency: str) -> dict:
    """Get cryptocurrency prices via CoinGecko API."""

使用模式：

User Query: "How much is 100 USD in GBP?"
         ↓
Agent calls get_exchange_rate("USD", "GBP")
         ↓
Returns: {"rate": 0.7479, "conversion": 74.79}
         ↓
Agent formats response: "100 USD = 74.79 GBP"

2. 天气服务

位置： mcp_weather_server/

@mcp.tool()
async def get_weather(city: str, country_code: str = "") -> dict:
    """Get current weather for any city using OpenWeatherMap API."""
    # Returns: temperature, conditions, humidity, wind speed, etc.

3. 网络搜索服务

位置： mcp_tavily_search/

@mcp.tool()
async def tavily_search(query: str, max_results: int = 5) -> list:
    """Search the web using Tavily AI search engine."""
    # Returns structured search results with summaries

4. YouTube 字幕提取器

位置： mcp_youtube_extraction_server/

@mcp.tool()
def extract_transcript(video_url: str) -> str:
    """Extract and return transcript from YouTube videos."""
    # Uses youtube-transcript-api
    # Returns full transcript with timestamps

5. 数学运算服务

位置： mcp_math_server/

@mcp.tool()
def calculate(expression: str) -> str:
    """Evaluate mathematical expressions safely."""
    # Supports complex math with numexpr library

6. Neo4j 图数据库服务

位置： neo4j-mcp-workspace-template/

用于图知识管理的五个专用 MCP 服务：

A. 数据建模（neo4j-data-modeling）

• 定义图架构
• 创建本体
• 管理实体类型和关系

B. 数据导入（neo4j-ingest）

• 从文件（CSV、JSON）加载数据
• 解析结构化/非结构化文档
• 创建节点和关系

C. 词汇图（neo4j-lexical-graph）

本项目中最复杂的实现

@mcp.tool()
async def create_lexical_graph(path: str) -> str:
    """
    Create a complete lexical graph from PDFs:
    1. Parse PDFs (docling, pymupdf)
    2. Extract elements (tables, images, text)
    3. Create nodes in Neo4j
    4. Build reading order chains
    """
    
    # Strategies: token_window, structured, by_section, by_page
    # Supports parallel processing via ProcessPoolExecutor
    # Includes VLM-based element description generation

详细工具集：

1. create_lexical_graph - 将 PDF 导入为结构化图
2. check_processing_status - 监控后台任务
3. cancel_processing - 停止正在运行的任务
4. chunk_lexical_graph - 将文档拆分为语义分块
5. assign_section_hierarchy - 创建文档大纲结构
6. generate_chunk_descriptions - 使用大模型描述分块
7. embed_chunks - 生成并存储嵌入
8. verify_lexical_graph - 验证图的完整性
9. reconstruct_document - 将图导出回 Markdown

D. 实体图（neo4j-entity-graph）

• 提取命名实体
• 创建实体关系
• 构建语义网络

E. GraphRAG

• 实现 RAG（检索增强生成）
• 查询图以获取上下文
• 根据图数据生成 AI 响应

7. 数据库服务

MySQL 服务

@mcp.tool()
async def execute_query(database: str, query: str) -> dict:
    """Execute SQL queries against MySQL database."""
    
@mcp.tool()
async def list_tables(database: str) -> list:
    """List all tables in a database."""

Snowflake 集成

@mcp.tool()
async def query_snowflake(sql: str) -> dict:
    """Execute queries on Snowflake data warehouse."""

AWS 集成

@mcp.tool()
async def list_s3_objects(bucket: str) -> list:
    """List objects in AWS S3 bucket."""
    
@mcp.tool()
async def get_rds_databases() -> list:
    """List RDS database instances."""

---

A2A（Agent-to-Agent）集成

什么是 A2A？

Agent-to-Agent（A2A）协议支持多个 Agent 协同工作的编排。它是 Google Agent 生态系统的一部分。

A2A 架构

┌──────────────┐
│ Client Apps  │
└────────┬─────┘
         │ A2A Protocol (JSON-RPC)
    ┌────▼──────────────────────────┐
    │   A2A Agent Executor          │
    │   (Runs on port 10030)        │
    │                               │
    │  ┌─────────────────────────┐  │
    │  │ Root Agent              │  │
    │  │ - Orchestrates tasks    │  │
    │  │ - Routes to sub-agents  │  │
    │  │ - Manages context       │  │
    │  └──────────────┬──────────┘  │
    │                 │             │
    │  ┌──────────────▼───────┐     │
    │  │ MCP Tools            │     │
    │  │ (Currency, Weather,  │     │
    │  │  Search, etc.)       │     │
    │  └──────────────────────┘     │
    └───────────────────────────────┘
         │
    ┌────▼────────────────────────────┐
    │ External Services               │
    │ (APIs, Databases, LLMs)         │
    └─────────────────────────────────┘

A2A 实现示例

文件： google_adk_agent_mcp/currency_agent.py

from google.adk import agent as agent_lib
from google.adk.a2a.utils.agent_to_a2a import to_a2a
from mcp_toolset import MCPToolset

# Step 1: Load MCP tools
mcp_tools = MCPToolset(
    server_url="http://localhost:7080/mcp"
).get_tools()

# Step 2: Define root agent
root_agent = agent_lib.Agent(
    instructions="Help users with currency conversion and financial queries",
    tools=[*mcp_tools, *other_tools],
    model="models/gemini-2.0-flash"
)

# Step 3: Convert to A2A
a2a_app = to_a2a(root_agent, port=10030)

# Step 4: Run server
if __name__ == "__main__":
    a2a_app.run()

A2A 客户端通信

文件： google_adk_agent_mcp/test_client.py

from google.adk.a2a.client import A2AClient
import json

# Connect to A2A agent
client = A2AClient("http://localhost:10030")

# Single-turn request
response = client.query(
    message="What is 100 USD in CAD?",
    context_id="session-123"
)

# Multi-turn conversation
context_id = "session-456"

# First turn
response1 = client.query("How much is 100 USD?", context_id)
# Agent: "Please specify target currency"

# Second turn (history is maintained)
response2 = client.query("in GBP", context_id)
# Agent: "100 USD = 74.79 GBP"

# Response structure:
{
    "id": "task-id",
    "status": "completed",
    "artifacts": [
        {
            "name": "conversion_result",
            "parts": [{"kind": "text", "text": "..."}]
        }
    ],
    "history": [
        {"role": "user", "text": "..."},
        {"role": "agent", "text": "..."}
    ]
}

多 Agent 编排

A2A 支持复杂工作流：

Task: Comprehensive financial analysis

1. Currency Converter Agent
   - Converts amounts between currencies
   - Calls: get_exchange_rate()

2. Crypto Analyst Agent
   - Analyzes cryptocurrency trends
   - Calls: get_crypto_prices()

3. Summary Agent
   - Combines insights
   - Calls: tavily_search() for market news

Client Request
    │
    ├─► A2A Router
    │   │
    │   ├─► Sub-Agent 1 (Currency) ──► MCP Currency Server
    │   │
    │   ├─► Sub-Agent 2 (Crypto) ────► MCP Crypto Server
    │   │
    │   └─► Sub-Agent 3 (Summary) ───► MCP Search Server
    │       │
    ├─ Aggregate Results
    └─► Return to Client

---

高级实现

1. 词汇图处理管道

Neo4j 词汇图服务展示了高级 MCP 模式：

处理流程

Input PDF Files
    │
    ├─ Pre-Flight Phase
    │  ├── Count files and pages
    │  ├── Estimate processing time
    │  └── Request user confirmation
    │
    ├─ Parse Phase (Async, Parallel)
    │  ├── docling (modern, multi-format)
    │  ├── pymupdf (fast, structured)
    │  ├── text_only (fallback)
    │  └── vlm_blocks (vision-based)
    │
    ├─ Element Extraction
    │  ├── Text blocks
    │  ├── Tables (preserved as structured data)
    │  ├── Images (base64 encoded)
    │  ├── Headings (hierarchical)
    │  └── Metadata
    │
    ├─ Neo4j Graph Creation
    │  ├── Create Document node
    │  ├── Create Page nodes
    │  ├── Create Element nodes
    │  ├── Link with relationships
    │  └── Build reading order NEXT chains
    │
    ├─ Chunking Phase
    │  ├── token_window (overlapping)
    │  ├── structured (by elements)
    │  ├── by_section (uses hierarchy)
    │  └── by_page (page-aware)
    │
    ├─ Enrichment Phase
    │  ├── Create section hierarchy
    │  ├── Generate descriptions (LLM)
    │  ├── Create embeddings
    │  └── Index for full-text search
    │
    ├─ Verification Phase
    │  ├── Check orphan nodes
    │  ├── Validate NEXT chains
    │  ├── Count statistics
    │  └── Reconstruct as Markdown
    │
    └─ Output
       ├── Success report
       ├── Processing logs
       ├── Reconstruction files
       └── Graph statistics

任务管理模式

class Job:
    job_id: str
    status: JobStatus  # PENDING, RUNNING, COMPLETE, FAILED
    progress: float    # 0.0 to 1.0
    elapsed: timedelta
    files_completed: int
    files_total: int
    
    def to_status_dict(self) -> dict:
        return {
            "job_id": self.job_id,
            "status": self.status.value,
            "progress_percentage": self.progress * 100,
            "elapsed_seconds": self.elapsed.total_seconds(),
            "files": f"{self.files_completed}/{self.files_total}",
            "eta_seconds": self.estimate_remaining()
        }

异步处理模式

@mcp.tool()
async def create_lexical_graph(path: str) -> str:
    # Create background job
    job = Job(job_id=uuid4(), status=JobStatus.PENDING)
    _job_manager.add_job(job)
    
    # Start processing in background
    asyncio.create_task(_process_async(job))
    
    # Return immediately with job ID
    return json.dumps({
        "status": "processing",
        "job_id": job.job_id,
        "message": "Processing started. Check status with check_processing_status()"
    })

async def _process_async(job: Job):
    try:
        job.status = JobStatus.RUNNING
        
        # Count files
        pdf_files = list(path.glob("*.pdf"))
        job.files_total = len(pdf_files)
        
        # Process each file
        for i, pdf in enumerate(pdf_files):
            await process_pdf(pdf, neo4j_driver, database)
            job.files_completed = i + 1
            job.progress = (i + 1) / job.files_total
            
        job.status = JobStatus.COMPLETE
    except Exception as e:
        job.status = JobStatus.FAILED
        job.error = str(e)

2. 文档重建

从图中重建文档的能力体现了图的完整性：

async def reconstruct_document(document_id: str, output_dir: str) -> str:
    """
    Reconstructs original document structure:
    
    From Graph:
    Document {id} ──HAS_PAGE──► Page {pageNumber}
                 ──HAS_ELEMENT─► Element {type, text, position}
    
    Creates reading-order chain via NEXT relationships
    
    Outputs:
    1. Element-based reconstruction (reading order)
    2. Chunk-based reconstruction (if chunks exist)
    3. Both as .md files for comparison
    """
    
    # Restore from graph
    elements = await get_elements_in_order(document_id)
    
    # Group by page
    pages = {}
    for element in elements:
        page_num = element.page_number
        if page_num not in pages:
            pages[page_num] = []
        pages[page_num].append(element)
    
    # Generate Markdown
    md_lines = []
    for page_num in sorted(pages.keys()):
        md_lines.append(f"# Page {page_num + 1}")
        for elem in pages[page_num]:
            if elem.type == "heading":
                md_lines.append(f"{'#' * (elem.level + 1)} {elem.text}")
            elif elem.type == "table":
                md_lines.append(f"\n**[TABLE]**\n{elem.text}\n")
            else:
                md_lines.append(elem.text)
    
    # Write output
    output_file.write_text("\n".join(md_lines))
    return f"Reconstruction saved to {output_file}"

3. 多框架集成

该项目展示了跨多个 AI 框架的集成：

┌──────────────────────────────────────────────────────┐
│          Integration Points                          │
├──────────────────────────────────────────────────────┤
│                                                      │
│  LangChain ◄────► MCP Servers                       │
│                  (via MultiServerMCPClient)         │
│                                                      │
│  Google ADK ◄──► A2A Protocol ◄─────► MCP Servers   │
│                  (via MCPToolset)                    │
│                                                      │
│  FastAPI ◄─────► FastMCP ◄─────► Transport          │
│                  (SSE bridge)      (Stdio/SSE)       │
│                                                      │
│  Claude Desktop ◄── Stdio ──► MCP Servers           │
│                  (stdio.json)                        │
│                                                      │
└──────────────────────────────────────────────────────┘

---

部署与配置

1. 本地开发环境设置

安装

# Clone or navigate to project
cd MCP_servers_and_a2a_utils

# Create virtual environment
python3.11 -m venv venv
source venv/bin/activate

# Install dependencies
pip install -r requirements.txt
pip install fastmcp mcp httpx

# For specific features:
pip install python-dotenv neo4j pymupdf docling-core
pip install google-cloud-aiplatform  # For Google ADK
pip install langchain langchain-mcp-adapters  # For LangChain integration

环境设置

# Create .zprofile or .env with:
export OPENAI_API_KEY="sk-..."
export TAVILY_API_KEY="tvly-..."
export OPENWEATHERMAP_API_KEY="..."
export NEO4J_URI="neo4j://localhost:7687"
export NEO4J_USERNAME="neo4j"
export NEO4J_PASSWORD="password"
export GOOGLE_APPLICATION_CREDENTIALS="/path/to/credentials.json"

2. 运行单个服务

FastMCP 货币服务

cd google_adk_agent_mcp
python -m uvicorn currency_agent:app --host 0.0.0.0 --port 7080

# Server starts with SSE at http://0.0.0.0:7080/sse

Neo4j 词汇图服务

cd neo4j-mcp-workspace-template/mcp-neo4j-lexical-graph

# Stdio transport (for Claude Desktop)
python -m mcp_neo4j_lexical_graph.server --transport stdio

# SSE transport (for HTTP clients)
python -m mcp_neo4j_lexical_graph.server --transport sse --host 0.0.0.0 --port 8002

天气服务

cd mcp_weather_server
python weather.py  # Starts on port 8000

3. Claude Desktop 配置

文件： claude_config.json

{
  "mcpServers": {
    "currency": {
      "command": "python",
      "args": ["google_adk_agent_mcp/currency_agent.py"],
      "transport": "stdio"
    },
    "weather": {
      "command": "python",
      "args": ["mcp_weather_server/weather.py"],
      "transport": "stdio"
    },
    "neo4j_lexical": {
      "command": "python",
      "args": ["neo4j-mcp-workspace-template/mcp-neo4j-lexical-graph/src/mcp_neo4j_lexical_graph/server.py"],
      "env": {
        "NEO4J_URI": "neo4j://localhost:7687",
        "NEO4J_USERNAME": "neo4j",
        "NEO4J_PASSWORD": "password"
      }
    }
  }
}

在 Claude Desktop 中使用：

1. 将 claude_config.json 复制到 ~/.claude/resources/mcp.json
2. 重启 Claude Desktop
3. Claude 即可使用所有服务提供的工具

4. MCP Inspector 调试

# Start inspector
npx @modelcontextprotocol/inspector

# Open in browser
http://127.0.0.1:6274/?MCP_PROXY_AUTH_TOKEN=<token>

# Test tools:
# - Select server
# - List available tools
# - Invoke with parameters
# - View results and errors

5. 运行 A2A Agent

# Start A2A server
cd google_adk_agent_mcp
python currency_agent.py

# Server runs on port 10030
# Logs show connection status

# In another terminal, run client
python test_client.py

# Creates conversation with agent

---

用例与示例

1. 货币兑换工作流

User: "How much is 100 USD in GBP?"
│
▼
Claude/Agent processes query
│
├─ Recognizes: Currency conversion needed
├─ Identifies: Source=USD, Target=GBP, Amount=100
│
▼
Calls MCP Tool: get_exchange_rate("USD", "GBP")
│
▼
A2A Agent (or Claude) receives tool result:
{
  "rate": 0.7479,
  "from": "USD",
  "to": "GBP",
  "conversion": 100 * 0.7479 = 74.79
}
│
▼
Formats response: "100 USD equals approximately 74.79 GBP"
│
▼
Returns to user

使用的组件：

• 货币兑换服务（MCP）
• A2A 协议层（可选）
• 工具调用引擎
• 响应格式化

2. 文档摄取与 RAG

Process: Upload Research Papers → Build Knowledge Graph → Query with RAG

Step 1: Ingest Document
┌──────────────────────────┐
│ User uploads: paper.pdf  │
└──────────────┬───────────┘
               │
               ▼
        ┌────────────────────┐
        │ Neo4j Lexical      │
        │ create_lexical_    │
        │ graph(pdf_path)    │
        └────────┬───────────┘
                 │
      Processing in background
                 │
        ┌────────▼──────────────┐
        │ Job Status            │
        │ - Parse PDF           │
        │ - Extract elements    │
        │ - Create nodes/edges  │
        │ - Build reading order │
        └────────┬──────────────┘
                 │
               Complete
                 │
        ┌────────▼──────────────┐
        │ Document Graph        │
        │ - Document node       │
        │ - Page nodes          │
        │ - Element nodes       │
        │ - Chunk nodes         │
        │ - Relationship indexes│
        └──────────────────────┘

Step 2: Query with RAG
┌──────────────────────┐
│ User Query: Explain  │
│ the paper's main     │
│ contribution         │
└──────────┬───────────┘
           │
           ▼
    ┌──────────────────────┐
    │ GraphRAG Server      │
    │ query_with_rag(...)  │
    └──────────┬───────────┘
               │
    ┌──────────▼──────────────────┐
    │ 1. Vector search in embeddings
    │ 2. Retrieve context chunks   │
    │ 3. Extract entity paths      │
    │ 4. Build prompt with context │
    └──────────┬──────────────────┘
               │
               ▼
    ┌──────────────────────┐
    │ LLM (Claude)         │
    │ Generate answer      │
    └──────────┬───────────┘
               │
               ▼
    Return RAG-augmented response

3. 多 Agent 协作

Task: Comprehensive Travel Planning

User Request: "Plan a trip to Japan with $5000 budget"
│
▼
A2A Router decides:
│
├─► Currency Agent
│   "Convert $5000 to JPY"
│   Calls: get_exchange_rate("USD", "JPY")
│   Returns: ≈750,000 JPY estimate
│
├─► Search Agent
│   "Find flights and hotels in Japan"
│   Calls: tavily_search("Japan flights hotels")
│   Returns: Recent deals and recommendations
│
├─► Weather Agent
│   "Check Japan climate for travel dates"
│   Calls: get_weather("Tokyo", "JP")
│   Returns: Seasonal weather patterns
│
└─► Aggregator Agent
    Combines all information:
    - Budget in local currency
    - Available options
    - Best seasons
    Returns: Comprehensive travel plan

Response to User:
"With $5000 (≈750,000 JPY), you can:
- Stay 10-14 days
- Visit 2-3 cities
- Best season: Spring/Fall
- Current flight deals available..."

4. 实时数据应用

加密货币投资组合跟踪器

# Runs as A2A agent
@tool
async def get_portfolio_summary(portfolio_dict: dict) -> str:
    """
    Tracks cryptocurrency holdings and provides real-time analysis
    
    portfolio_dict = {
        "bitcoin": 0.5,
        "ethereum": 5.0,
        "cardano": 1000.0
    }
    """
    
    prices = {}
    total_value = 0
    
    for symbol, amount in portfolio_dict.items():
        price = await get_crypto_prices(symbol, "usd")
        value = amount * price
        prices[symbol] = {
            "amount": amount,
            "price_per_unit": price,
            "total_value": value
        }
        total_value += value
    
    return format_portfolio_report(prices, total_value)

---

最佳实践与经验教训

1. MCP 服务设计

✅ 应做事项

• 使用 Pydantic 模型提供强类型提示
• 为所有工具提供详细的文档字符串
• 添加工具注释，让客户端清楚了解工具
• 使用 ToolError 优雅地处理错误
• 支持异步操作
• 验证输入参数
• 返回结构化且可解析的结果

❌ 不应做的事项

• 不要假定客户端会处理错误
• 不要暴露系统实现细节
• 不要创建需要大量设置的工具
• 不要向用户返回原始异常消息
• 不要在没有超时设置的情况下执行阻塞操作
• 不要在工具描述中暴露敏感凭据

示例：规范的工具设计

from pydantic import Field
from mcp.types import ToolError
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("ExampleServer")

@mcp.tool()
async def safe_operation(
    parameter: str = Field(
        ...,
        description="Clear description of what this parameter does",
        json_schema_extra={"example": "good_value"}
    ),
    timeout: int = Field(
        30,
        description="Maximum seconds to wait for operation"
    )
) -> dict:
    """
    Descriptive tool docstring explaining:
    - What the tool does
    - When to use it
    - What it returns
    - Possible error conditions
    """
    try:
        # Validate input
        if not parameter:
            raise ToolError("Parameter cannot be empty")
        
        if timeout < 1 or timeout > 300:
            raise ToolError("Timeout must be between 1 and 300 seconds")
        
        # Perform operation
        result = await perform_operation(parameter, timeout=timeout)
        
        # Return structured result
        return {
            "status": "success",
            "result": result,
            "metadata": {"operation": "safe_operation"}
        }
        
    except ToolError:
        # Re-raise known errors
        raise
    except Exception as e:
        # Log and convert unknown errors
        logger.error(f"Operation failed: {e}")
        raise ToolError(f"Operation failed: {str(e)}")

2. A2A Agent 编排

Agent 设计模式

# Pattern 1: Hierarchical Agents
root_agent
├── currency_sub_agent
│   └── Tools: exchange_rate, crypto_prices
├── search_sub_agent
│   └── Tools: web_search, news
└── analysis_sub_agent
    └── Tools: sentiment_analysis, trend_detection

# Pattern 2: Tool-Sharing Agents
agent_a ─┬─► shared_tool_1
         ├─► shared_tool_2
agent_b ─┘   (reduces duplication)

# Pattern 3: Fan-Out-Fan-In
root_agent
├─ Task 1 → agent_1 → task_1a, task_1b (parallel)
├─ Task 2 → agent_2 → task_2a, task_2b (parallel)
└─ Combine results

3. 错误处理模式

# Pattern 1: Graceful Degradation
@mcp.tool()
async def get_data_with_fallback(primary_source: str) -> dict:
    try:
        return await fetch_from_primary(primary_source)
    except ConnectionError:
        logger.warning(f"Primary source {primary_source} unavailable")
        return await fetch_from_fallback()
    except TimeoutError:
        logger.warning("Operation timed out, returning cached data")
        return get_cached_data()

# Pattern 2: Structured Error Responses
@mcp.tool()
async def safe_tool(param: str) -> dict:
    try:
        result = await process(param)
        return {
            "status": "success",
            "data": result
        }
    except ValidationError as e:
        raise ToolError(f"Invalid input: {e.message}")
    except PermissionError as e:
        raise ToolError(f"Permission denied: {e.message}")
    except Exception as e:
        logger.error(f"Unexpected error: {e}", exc_info=True)
        raise ToolError(f"Internal error occurred")

4. 性能优化

异步/并发模式

# ❌ Slow: Sequential processing
for pdf_file in pdf_files:
    await process_pdf(pdf_file)  # Processes one at a time

# ✅ Fast: Parallel processing
tasks = [process_pdf(pdf) for pdf in pdf_files]
results = await asyncio.gather(*tasks, return_exceptions=True)

# ✅ Better: Bounded concurrency
semaphore = asyncio.Semaphore(max_workers)
async def process_bounded(pdf):
    async with semaphore:
        return await process_pdf(pdf)

tasks = [process_bounded(pdf) for pdf in pdf_files]
results = await asyncio.gather(*tasks)

# ✅ Best for CPU-bound: Process pool
from concurrent.futures import ProcessPoolExecutor

with ProcessPoolExecutor(max_workers=4) as pool:
    results = await asyncio.get_event_loop().run_in_executor(
        pool, cpu_intensive_function, argument
    )

数据库连接池

# Neo4j connection pooling (automatic)
from neo4j import AsyncGraphDatabase

driver = AsyncGraphDatabase.driver(
    "neo4j://localhost:7687",
    auth=("neo4j", "password"),
    max_connection_pool_size=100,
    connection_timeout=30.0
)

# Properly close when shutting down
await driver.close()

5. 测试 MCP 服务

# Test tool invocation
@pytest.mark.asyncio
async def test_get_exchange_rate():
    mcp = create_mcp_server()
    
    result = await mcp._tools["get_exchange_rate"].fn(
        currency_from="USD",
        currency_to="GBP"
    )
    
    assert result["status"] == "success"
    assert "rate" in result
    assert result["rate"] > 0

# Test error handling
@pytest.mark.asyncio
async def test_invalid_currency():
    mcp = create_mcp_server()
    
    with pytest.raises(ToolError):
        await mcp._tools["get_exchange_rate"].fn(
            currency_from="INVALID",
            currency_to="GBP"
        )

# Integration test with MCP Inspector
def test_server_startup():
    server = subprocess.Popen(
        ["python", "server.py"],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )
    
    time.sleep(2)  # Wait for server
    
    response = requests.get("http://localhost:8000/mcp")
    assert response.status_code == 200
    
    server.terminate()

6. 文档最佳实践

每个工具都应说明：

@mcp.tool()
async def comprehensive_tool(
    required_param: str = Field(
        ...,
        description="What this parameter does"
    ),
    optional_param: int = Field(
        42,
        description="What this controls",
        ge=1, le=100  # Validation
    )
) -> dict:
    """
    [1-2 sentence summary]
    
    [Detailed explanation of functionality]
    
    Parameters:
    - required_param: [Description with examples]
    - optional_param: [Description with range/options]
    
    Returns:
    {
        "status": "success" or "error",
        "data": [Actual return data structure],
        "metadata": [Optional metadata]
    }
    
    Raises:
    - ToolError: When [specific condition]
    
    Examples:
    >>> result = await comprehensive_tool("value", optional_param=50)
    >>> print(result["data"])
    """

---

高级主题

1. 自定义传输协议

虽然 Stdio 和 SSE 是标准协议，但你可以实现自定义传输方式：

from mcp.server import Server
from mcp.server.base import *

class CustomTransport(ServerTransport):
    """Custom transport implementation."""
    
    async def read(self) -> ServerMessage:
        """Read message from source."""
        pass
    
    async def write(self, message: ServerMessage) -> None:
        """Write message to destination."""
        pass

# Use custom transport
await mcp_server.run(CustomTransport())

2. 资源流式传输

对于大型资源，可实现流式传输：

@mcp.resource("data://large-dataset/{id}")
async def stream_dataset(id: str):
    # Stream data in chunks instead of loading all at once
    async def generate():
        async for chunk in fetch_chunks(id):
            yield chunk
    
    return {
        "uri": f"data://large-dataset/{id}",
        "mimeType": "application/octet-stream",
        "contents": [
            {"type": "text", "text": chunk} async for chunk in generate()
        ]
    }

3. LLM 采样（实验性）

MCP 规范包含用于生成文本的采样支持：

@server.sampling()
async def sample_text(
    messages: list,
    model: str = "claude-3-sonnet",
    max_tokens: int = 1024
) -> SamplingMessage:
    """Use Claude to generate text within MCP context."""
    # Implementation uses Claude API

4. 监控与可观测性

import logging
from functools import wraps

logger = logging.getLogger(__name__)

def with_metrics(func):
    """Decorator to track tool execution metrics."""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        import time
        start = time.time()
        
        try:
            result = await func(*args, **kwargs)
            duration = time.time() - start
            
            logger.info(
                "Tool executed",
                extra={
                    "tool": func.__name__,
                    "duration_ms": duration * 1000,
                    "status": "success"
                }
            )
            return result
        except Exception as e:
            duration = time.time() - start
            logger.error(
                "Tool failed",
                extra={
                    "tool": func.__name__,
                    "duration_ms": duration * 1000,
                    "error": str(e)
                }
            )
            raise
    
    return wrapper

@mcp.tool()
@with_metrics
async def monitored_tool():
    """Tool with automatic metrics collection."""
    pass

---

故障排除指南

常见问题与解决方案

问题	原因	解决方案
Inspector 中显示“Connection refused”	服务未运行	先使用 python server.py 启动服务
工具未出现在列表中	导入错误或未应用装饰器	检查 @mcp.tool() 装饰器，并确认导入语句
A2A 处于“Input required”状态	Agent 正在等待澄清	客户端应发送后续消息
Neo4j 连接超时	数据库无法访问	验证 NEO4J_URI、用户名和密码
PDF 处理时内存不足	并行处理的大型 PDF 过多	调低 max_parallel 参数
SSE 流意外关闭	网络超时或服务崩溃	检查服务日志并增大超时设置
工具结果无法序列化	数据类型与 JSON 不兼容	转换为字符串或结构化格式

---

结论

这个项目展现了一个全面、可用于生产环境的生态系统，可用于构建 AI Agent，具备以下特点：

1. 灵活的工具集成 - 20+ 个专用 MCP 服务
2. 支持多种框架 - LangChain、Google ADK、FastAPI
3. 精密的编排能力 - 通过 A2A 协议实现 Agent 协作
4. 高级处理能力 - 用于文档理解的 ML 流水线
5. 真实场景示例 - 货币、天气、搜索、RAG 系统

该架构展示了如何通过标准化协议连接 AI 模型与外部能力，从而实现复杂的多步骤工作流，同时保持清晰的抽象和关注点分离。

要点总结：

• MCP 与协议无关 - 适用于任何 AI 模型/客户端
• 异步/并发模式至关重要 - 有助于提升性能和可扩展性
• 类型安全很重要 - Pydantic 验证可及早防止错误
• 文档至关重要 - 工具描述有助于模型做出更好的决策
• 错误处理有助于建立信任 - 优雅降级可提升可靠性
• 可观测性有助于调试 - 日志记录和监控必不可少

对于任何使用工具、RAG 或多 Agent 编排来构建 AI 驱动系统的人来说，该项目都是一个优秀的参考实现。

安装步骤

1. pip install uv
2. uv init MCP_servers_and_a2a_utils
3. uv run mcp install main.py
4. uv run mcp

参考资料

1. https://medium.com/data-and-beyond/the-model-context-protocol-mcp-the-ultimate-guide-c40539e2a8e7
2. https://github.com/coleam00/ottomator-agents/blob/main/nba-agent/nba_agent.py
3. https://www.youtube.com/watch?v=v_6EXt6T83I
4. https://modelcontextprotocol.io/introduction
5. https://github.com/coleam00/ottomator-agents/tree/main/pydantic-ai-mcp-agent

示例参考

1. https://www.dremio.com/blog/building-a-basic-mcp-server-with-python/
2. https://github.com/HeetVekariya/Linear-Regression-MCP
3. https://github.com/singlestore-labs/mcp-server-singlestore
4. https://github.com/jlowin/fastmcp
5. https://github.com/awslabs/mcp/blob/main/src/aws-documentation-mcp-server/awslabs/aws_documentation_mcp_server/server.py

MCP 工作坊

1. https://github.com/akranga/mcp-workshop