MCP 服务器

Model Context Protocol (MCP) Example

模型上下文协议（MCP）的参考实现，支持LLM和应用程序之间的无缝工具调用。其特点是客户端/服务器架构，具有HTTP API、本地CLI执行以及AWS Bedrock集成，并且是一个生产就绪、可扩展的框架。

iddv

开发者工具

访问服务器

README

模型上下文协议 (MCP) 示例

LLM 和应用程序之间进行工具调用的模型上下文协议 (MCP) 的参考实现。

概述

本项目既是模型上下文协议的客户端实现，也是服务端实现。它演示了：

通过 stdio 进行本地工具调用
通过 HTTP 进行远程工具调用
与 AWS Bedrock 和 Claude 3.7 集成

模型上下文协议通过定义函数定义、函数调用和响应的标准格式，促进 LLM 与工具或应用程序之间的交互。

AI 辅助开发

本项目是在 Anthropic 的高级 AI 助手 Claude 3.7 的协助下创建的。规划、实施策略和技术决策记录在 llm/ 目录中，该目录包含：

设计规范
实施计划
技术决策
开发进度

这些文档提供了对 AI 辅助开发过程的深入了解，可以作为类似项目的参考。

项目结构

mcp-example/
├── core/                 # 核心协议实现
│   ├── schema.py         # 协议模式定义
│   ├── validation.py     # 模式验证实用程序
│   ├── registry.py       # 工具注册表
│   └── executor.py       # 工具执行器
├── tools/                # 工具实现
│   ├── calculator.py     # 计算器工具
│   └── text.py           # 文本处理工具
├── adapters/             # 接口适配器
│   ├── stdio/            # 命令行界面
│   └── http/             # 远程服务器的 HTTP 客户端
├── server/               # 服务器实现
│   ├── app.py            # FastAPI 服务器
│   └── main.py           # 服务器运行器
├── examples/             # 用法示例
├── tests/                # 测试套件
└── llm/                  # 实现文档

快速入门

前提条件

Python 3.10 或更高版本
Poetry (用于依赖管理)

安装

克隆存储库：

git clone https://github.com/yourusername/mcp-example.git
cd mcp-example

设置虚拟环境并安装依赖项：

选项 1：使用 Poetry（推荐）：

poetry install

选项 2：使用 venv：

python3 -m venv venv
source venv/bin/activate  # 在 Windows 上，使用：venv\Scripts\activate
pip install -e .

运行 CLI

命令行界面提供了一种在本地与工具交互的方式：

# 使用 Poetry
poetry run python -m mcp_example.adapters.stdio.cli

# 使用 venv
python -m mcp_example.adapters.stdio.cli

这将启动一个 REPL，您可以在其中：

使用 list 命令列出可用工具
使用 JSON 语法直接调用工具：{"name": "calculator", "parameters": {"operation": "add", "a": 5, "b": 3}}
使用 help 命令获取帮助

运行服务器

FastAPI 服务器为工具调用提供了一个远程 API：

# 使用 Poetry
poetry run python -m mcp_example.server.main --host 0.0.0.0 --port 8000

# 使用 venv
python -m mcp_example.server.main --host 0.0.0.0 --port 8000

默认情况下，这会在 http://127.0.0.1:8000 上启动一个服务器。您可以在 http://127.0.0.1:8000/docs 访问 API 文档。

服务器选项：

--host: 绑定到的主机 (默认: 127.0.0.1)
--port: 监听的端口 (默认: 8000)
--reload: 启用自动重新加载以进行开发
--log-level: 设置日志级别 (debug, info, warning, error)

测试服务器

服务器运行后，您可以使用 curl 进行测试：

列出可用函数：

curl -X GET http://localhost:8000/api/functions -H "X-API-Key: test-key"

调用计算器函数：

curl -X POST http://localhost:8000/api/functions/call \
  -H "X-API-Key: test-key" \
  -H "Content-Type: application/json" \
  -d '{"name": "calculator", "parameters": {"operation": "add", "a": 5, "b": 3}}'

将文本转换为大写：

curl -X POST http://localhost:8000/api/functions/call \
  -H "X-API-Key: test-key" \
  -H "Content-Type: application/json" \
  -d '{"name": "transform_text", "parameters": {"operation": "uppercase", "text": "hello world"}}'

分析文本：

curl -X POST http://localhost:8000/api/functions/call \
  -H "X-API-Key: test-key" \
  -H "Content-Type: application/json" \
  -d '{"name": "analyze_text", "parameters": {"text": "Hello world. This is a test."}}'

故障排除

如果您遇到任何问题：

确保已安装所有依赖项：

pip list | grep uvicorn  # 应该显示已安装 uvicorn

检查日志中是否存在循环导入错误：

python -m mcp_example.server.main --log-level debug

验证您的请求中是否包含 API 密钥（默认为 "test-key"）

API 端点

服务器提供以下端点：

GET /api/functions: 列出所有可用函数
GET /api/functions/{name}: 获取特定函数定义
POST /api/functions/call: 调用函数
POST /api/tools/call: 调用工具
POST /api/execute: 从文本执行函数调用
WebSocket /api/functions/stream: 流式传输函数结果
WebSocket /api/tools/stream: 流式传输工具结果

使用 HTTP 客户端

要从 Python 应用程序调用服务器：

from mcp_example.adapters.http.client import MCPClient

# 创建客户端
client = MCPClient(
    base_url="http://localhost:8000",
    api_key="test-key"  # 使用默认的测试密钥
)

# 列出可用函数
functions = client.list_functions()
for func in functions:
    print(f"{func.name}: {func.description}")

# 调用函数
response = client.call_function(
    name="calculator",
    parameters={"operation": "add", "a": 5, "b": 3}
)
print(f"Result: {response.result}")

WebSocket 流式传输

MCP 实现支持使用 WebSockets 流式传输长时间运行操作的结果。这对于以下情况特别有用：

生成增量结果的函数
进度更新有价值的长时间运行的操作
需要显示部分结果的实时用户界面

AsyncMCPClient 提供了用于流式传输函数和工具结果的方法：

import asyncio
from mcp_example.adapters.http.client import AsyncMCPClient

async def main():
    # 创建异步客户端
    client = AsyncMCPClient("http://localhost:8000", api_key="test-key")
    
    # 从长时间运行的函数流式传输结果
    print("Streaming function results:")
    async for chunk in client.stream_function(
        name="long_running_operation",
        parameters={"duration": 5}
    ):
        # 逐块处理
        if chunk.status == "in_progress":
            print(f"Progress: {chunk.result}")
        elif chunk.status == "complete":
            print(f"Final result: {chunk.result}")
        elif chunk.status == "error":
            print(f"Error: {chunk.error}")
    
    await client.close()

if __name__ == "__main__":
    asyncio.run(main())

每个流式传输块包含：

id: 块的唯一标识符
status: 操作的状态 ("in_progress", "complete", 或 "error")
result: 部分或最终结果数据
error: 如果状态为 "error" 的错误信息
timestamp: 创建块的时间

工具结果缓存

HTTP 客户端支持缓存工具和函数结果，以提高性能并减少冗余网络调用。这对于幂等操作或使用相同参数重复调用同一工具时特别有用。

要将缓存与 HTTP 客户端一起使用：

# 创建具有缓存选项的客户端
client = MCPClient(
    base_url="http://localhost:8000",
    api_key="test-key",
    cache_enabled=True,      # 启用/禁用缓存 (默认: True)
    cache_max_size=100,      # 最大缓存条目数 (默认: 100)
    cache_ttl=300.0          # 缓存生存时间（以秒为单位）(默认: 300.0)
)

# 第一次调用将命中服务器
result1 = client.call_function("calculator.add", {"a": 1, "b": 2})

# 使用相同参数的第二次调用将使用缓存的结果
result2 = client.call_function("calculator.add", {"a": 1, "b": 2})

# 绕过特定调用的缓存
result3 = client.call_function("calculator.add", {"a": 1, "b": 2}, use_cache=False)

# 使特定缓存条目无效
client.invalidate_cache_entry("calculator.add", {"a": 1, "b": 2})

# 清除整个缓存
client.clear_cache()

缓存行为：

达到最大大小时，使用 LRU（最近最少使用）驱逐
条目在配置的 TTL 后自动过期
参数已标准化，以便无论顺序如何，都能保持一致的缓存
可以全局或为单个请求禁用缓存

AWS Bedrock 和 Claude 3.7 集成

MCP 实现包括与 AWS Bedrock 的集成，特别是与 Claude 3.7 的集成。这使您可以利用 Claude 的高级自然语言理解和函数调用功能，同时使用标准 MCP 工具。

设置 AWS Bedrock

创建 AWS 账户并启用 Bedrock：
- 如果您没有账户，请注册 AWS
- 导航到 AWS Bedrock 服务
- 请求访问 Claude 3.7 模型（可能需要批准）

配置 AWS 凭证：

export AWS_ACCESS_KEY_ID="your-access-key"
export AWS_SECRET_ACCESS_KEY="your-secret-key"
export AWS_DEFAULT_REGION="us-west-2"  # 或您首选的区域

将 Claude 3.7 与 MCP 结合使用

基本用法示例：

from mcp_example.adapters.aws.claude import ClaudeAdapter, ClaudeMessage, ClaudeRole
from mcp_example.core.schema import FunctionDefinition

# 创建 Claude 适配器
adapter = ClaudeAdapter()

# 为 Claude 创建消息
messages = [
    ClaudeMessage(role=ClaudeRole.USER, content="What's 42 + 7?")
]

# 定义 Claude 可以调用的计算器函数
calculator_function = FunctionDefinition(
    name="calculator",
    description="Performs arithmetic operations",
    parameters={
        "type": "object",
        "properties": {
            "operation": {
                "type": "string",
                "enum": ["add", "subtract", "multiply", "divide"],
                "description": "The operation to perform"
            },
            "a": {"type": "number", "description": "First operand"},
            "b": {"type": "number", "description": "Second operand"}
        },
        "required": ["operation", "a", "b"]
    }
)

# 生成具有函数调用的响应
response = adapter.generate(messages, functions=[calculator_function])

# 提取函数调用
function_calls = adapter.extract_function_calls(response)
for call in function_calls:
    print(f"Function: {call.name}, Parameters: {call.parameters}")

交互式对话示例

您可以运行与 Claude 3.7 的交互式对话，该对话使用工具，如下所示：

import asyncio
from mcp_example.examples.claude_conversation import conversation_with_tools

# 定义用户消息
messages = [
    "What's 128 divided by 4?",
    "Convert 'hello world' to uppercase.",
    "What is the square root of 144?"
]

# 运行启用流式传输的对话
asyncio.run(conversation_with_tools(messages, streaming=True))

这将产生一个交互式对话，其中 Claude 使用适当的工具来响应用户查询。

使用 Claude 进行流式传输

Claude 3.7 支持流式传输响应，这对于长篇内容尤其有用：

from mcp_example.adapters.aws.claude import AsyncClaudeAdapter, ClaudeMessage, ClaudeRole

# 创建异步适配器
adapter = AsyncClaudeAdapter()

# 创建消息
messages = [
    ClaudeMessage(role=ClaudeRole.USER, content="Explain quantum computing in simple terms.")
]

# 流式传输响应
async for chunk in adapter.generate_with_streaming(messages):
    if chunk.content:
        print(chunk.content, end="", flush=True)

有关 AWS 设置和 Claude 集成的更多详细信息，请参阅 llm/ 目录中的文档：

aws_setup.md: 详细的 AWS 配置说明
claude_integration.md: Claude API 集成详细信息
claude_conversation.md: 对话流程实现

代理工具

代理工具可以转发函数调用到远程 MCP 服务器，从而可以访问其他服务器上可用的工具，而无需直接实现它们：

from mcp_example.core.executor import execute
from mcp_example.tools.proxy import register

# 确保已注册代理工具
register()

# 在远程服务器上调用函数
result = execute({
    "name": "proxy",
    "parameters": {
        "server_url": "http://remote-server.com",
        "api_key": "remote-server-key", 
        "function_name": "calculator",
        "parameters": {
            "operation": "add",
            "a": 5,
            "b": 3
        }
    }
})
print(f"Result from remote server: {result}")

从 CLI，您可以使用：

MCP> proxy server_url=http://remote-server.com function_name=calculator parameters={"operation":"add","a":5,"b":3}

工具链

MCP 实现支持将工具链接在一起以创建更复杂的工作流程。有几种工具链方法：

直接链接

from mcp_example.core.executor import execute

# 步骤 1：调用计算器工具
calc_result = execute({
    "name": "calculator",
    "parameters": {
        "operation": "add",
        "a": 5,
        "b": 3
    }
})

# 步骤 2：在文本工具中使用结果
text_result = execute({
    "name": "transform_text",
    "parameters": {
        "operation": "append",
        "text": "The result is: ",
        "append_text": str(calc_result)
    }
})

print(text_result)  # 输出: "The result is: 8"

基于代理的链接

from mcp_example.core.executor import execute
from mcp_example.tools.proxy import register

register()  # 确保已注册代理工具

# 步骤 1：使用代理工具调用远程函数
remote_result = execute({
    "name": "proxy",
    "parameters": {
        "server_url": "http://remote-server.com",
        "function_name": "data_service",
        "parameters": {"query": "get_user_info", "user_id": "12345"}
    }
})

# 步骤 2：在本地处理远程结果
processed_result = execute({
    "name": "transform_text",
    "parameters": {
        "operation": "extract",
        "text": remote_result["user_data"]["description"],
        "pattern": r"email: ([\w\.-]+@[\w\.-]+)"
    }
})

print(f"Extracted email: {processed_result}")

可用工具

计算器

执行基本算术运算：

add: 加法 (a + b)
subtract: 减法 (a - b)
multiply: 乘法 (a * b)
divide: 除法 (a / b)
power: 指数 (a ^ b)
sqrt: 平方根 (√a)
log: 对数 (log_b(a))

文本处理

提供文本转换和分析：

转换操作：

uppercase: 将文本转换为大写
lowercase: 将文本转换为小写
capitalize: 首字母大写
title: 转换为标题大小写
reverse: 反转文本
count_chars: 统计字符数
count_words: 统计字数
trim: 删除前导/尾随空格
replace: 替换文本

文本分析：

提供有关文本的统计信息（字符数、字数等）

贡献

Fork 存储库
创建一个功能分支
提交 pull request

许可证

本项目根据 MIT 许可证获得许可 - 有关详细信息，请参阅 LICENSE 文件。