AI导哈

计算器MCP服务

一个基于Model Context Protocol (MCP)的计算器服务，支持STDIO和SSE/HTTP两种通信协议，用于AI助手集成。

  • by Aksshay88
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 Aksshay88/MCP

使用说明

MCP计算器-技术文档

模型上下文协议（MCP）的全面实现，展示了用于AI辅助集成的STDIO（本地进程通信）和SSE/HTTP（网络通信）传输机制。

概述

该存储库包含计算器MCP服务器的两个并行实现，该服务器将算术运算作为Claude等AI助手的工具。这些实现展示了不同的传输协议，同时保持了相同的功能。

建筑

传输协议

STDIO（标准输入/输出）

  • 位置: STDIO/fastmcp_calculator.py
  • 协议：通过stdin/stdout进行本地进程通信
  • 用例：与本地AI客户端（Claude Desktop、IDE）直接集成
  • 连接：通过标准流进行IPC
  • 生命周期：基于进程，由宿主应用程序生成

通过HTTP的SSE（服务器发送事件）

  • 位置: SSE/fastapi-mcp_calculator.py
  • 协议：基于HTTP的SSE实时更新
  • 用例：网络可访问的MCP服务器，远程集成
  • 连接：HTTP端点位于 http://localhost:8002
  • 生命周期：长时间运行的web服务器进程

技术实现

STDIO实施

# STDIO/fastmcp_calculator.py
from fastmcp import FastMCP
mcp = FastMCP(name="Calculator")

主要特征：

  • 用途 FastMCP 简化MCP协议实现库
  • 以阻塞模式运行 mcp.run() -在stdin/stdout上监听
  • 通过管道进行同步和双向通信
  • 无需网络堆栈
  • JSON-RPC消息的自动序列化/反序列化
  • 进程由MCP主机（例如Claude Desktop）生成和管理

消息流：

Host Application (Claude Desktop)
    ↓ spawn process
Calculator MCP Server
    ↓ JSON-RPC via stdin
[Tool Request: add(5, 3)]
    ↓ stdout
[Tool Response: {"result": 8}]

SSE/HTTP实现

# SSE/fastapi-mcp_calculator.py
from fastapi import FastAPI
from fastapi_mcp import FastApiMCP

app = FastAPI(title="Calculator API")
mcp = FastApiMCP(app, name="Calculator MCP")
mcp.mount_http()

主要特征：

  • 基于FastAPI构建，具有强大的HTTP服务器功能
  • FastApiMCP 将FastAPI端点包装为MCP工具
  • 公开标准REST API和兼容MCP的端点
  • 支持跨源请求的CORS
  • 服务器发送流式响应事件
  • 在端口8002上与Uvicorn ASGI服务器一起运行
  • 支持来自多个客户端的并发连接

消息流：

HTTP Client/MCP Host
    ↓ HTTP POST
http://localhost:8002/multiply
    ↓ JSON payload
{"a": 5, "b": 3}
    ↓ HTTP Response
{"result": 15}

MCP端点结构：

POST /mcp/v1/tools/list    - List available tools
POST /mcp/v1/tools/call    - Execute a tool
GET  /mcp/v1/sse           - SSE endpoint for streaming

工具实施

这两种实现都公开了相同的计算器操作：

1.补充(add)

  • 函数: add_numbers(x: float, y: float) -> float
  • 描述：将两个数字相加
  • 标签: math, arithmetic （仅限STDIO）
  • 退货：x和y之和

2.乘法(multiply)

  • 函数: multiply(a: float, b: float) -> float
  • 描述：将两个数字相乘
  • 退货：a和b的乘积

3.减法(subtract)

  • 函数: subtract(a: float, b: float) -> float
  • 描述：从a中减去b
  • 退货：差异（a-b）

4.分部(divide)

  • 函数: divide(a: float, b: float) -> float
  • 描述：将a除以b
  • 错误处理:
    • STDIO：加薪 ValueError 除以零
    • SSE：回报 {"error": "Division by zero is not allowed."}
  • 退货：商数（a/b）

协议比较

功能 STDIO SSE/HTTP
运输 stdin/stdout管道 HTTP/1.1+SSE
网络 仅限本地 网络可访问
并发 单客户端 多客户端
安全 进程隔离 需要网络身份验证
延迟 ~1-5ms ~10-50ms（网络开销）
发现 进程生成 HTTP服务发现
状态管理 进程内存 服务器内存+会话
错误处理 Python异常 HTTP状态代码+JSON错误
调试 进程日志、stderr HTTP日志、请求跟踪

设置和安装

先决条件

Python 3.8+
pip package manager

STDIO设置

cd STDIO
python -m venv meraenv
source meraenv/bin/activate  # On Windows: meraenv\Scripts\activate
pip install fastmcp
python fastmcp_calculator.py

SSE/HTTP设置

cd SSE
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
pip install fastapi fastapi-mcp uvicorn
python fastapi-mcp_calculator.py

集成示例

克劳德桌面配置（STDIO）

增添 claude_desktop_config.json:

{
  "mcpServers": {
    "calculator": {
      "command": "python",
      "args": ["/absolute/path/to/STDIO/fastmcp_calculator.py"],
      "env": {}
    }
  }
}

HTTP客户端集成（SSE）

import requests

# Direct API call
response = requests.post(
    "http://localhost:8002/add",
    json={"a": 10, "b": 20}
)
print(response.json())  # {"result": 30}

# MCP protocol call
response = requests.post(
    "http://localhost:8002/mcp/v1/tools/call",
    json={
        "tool": "add",
        "arguments": {"a": 10, "b": 20}
    }
)

Types/JavaScript客户端

// SSE connection for streaming
const eventSource = new EventSource('http://localhost:8002/mcp/v1/sse');

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log('Received:', data);
};

// Tool invocation
const response = await fetch('http://localhost:8002/mcp/v1/tools/call', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({
    tool: 'multiply',
    arguments: { a: 7, b: 6 }
  })
});

协议深潜

MCP消息结构

工具列表请求：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list",
  "params": {}
}

工具列表响应：

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "tools": [
      {
        "name": "add",
        "description": "Add two numbers.",
        "inputSchema": {
          "type": "object",
          "properties": {
            "x": {"type": "number"},
            "y": {"type": "number"}
          },
          "required": ["x", "y"]
        }
      }
    ]
  }
}

工具调用请求：

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/call",
  "params": {
    "name": "add",
    "arguments": {
      "x": 5,
      "y": 3
    }
  }
}

工具调用响应：

{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "8"
      }
    ]
  }
}

性能注意事项

工作室

  • 优点：开销最小，JSON-RPC之外无序列化，直接内存访问
  • 缺点：单客户端限制，进程生成开销（~50-100ms）
  • 最适合：桌面应用程序、本地AI助手、开发环境

SSE/HTTP

  • 优点：可扩展、多个并发客户端、RESTful接口、服务发现
  • 缺点：网络延迟、HTTP开销、需要端口管理
  • 最适合：生产部署、云服务、分布式系统、微服务

安全考虑

工作室

  • 以与主机进程相同的权限运行
  • 无网络暴露
  • 流程执行时的信任边界
  • 建议：验证主机进程标识

SSE/HTTP

  • 暴露在网络接口上
  • 需要身份验证/授权层（示例中未实现）
  • 在没有TLS的情况下容易受到网络攻击
  • 建议：添加JWT身份验证，实现HTTPS，使用速率限制

调试和故障排除

标准调试模式

# Enable stderr logging
python fastmcp_calculator.py 2> debug.log

# Monitor JSON-RPC messages
strace -e read,write -s 1000 python fastmcp_calculator.py

HTTP调试模式

# Enable FastAPI debug mode
uvicorn.run(app, host="localhost", port=8002, log_level="debug")

# Test with curl
curl -X POST http://localhost:8002/add \
  -H "Content-Type: application/json" \
  -d '{"a": 5, "b": 3}'

常见问题

工作室：

  • 进程未生成：检查配置中的Python路径
  • 无响应：验证stdin/stdout未被其他输出阻止
  • 编码错误：确保两侧均采用UTF-8编码

HTTP：

  • 正在使用的端口：更改端口或终止现有进程
  • 连接被拒绝：验证服务器是否正在运行 lsof -i :8002
  • CORS错误：为浏览器客户端添加CORS中间件

扩大实施范围

添加新工具（STDIO）

@mcp.tool(
    name="power",
    description="Raise a number to a power",
    tags={"math", "advanced"}
)
def power(base: float, exponent: float) -> float:
    """Calculate base^exponent"""
    return base ** exponent

添加新端点（HTTP）

@app.post("/power")
def power_numbers(base: float, exponent: float):
    """Raise base to the power of exponent"""
    result = base ** exponent
    return {"result": result}

测试

单元测试

# test_calculator.py
def test_add():
    assert add_numbers(5, 3) == 8

def test_divide_by_zero():
    with pytest.raises(ValueError):
        divide(10, 0)

集成测试

# STDIO
echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | python fastmcp_calculator.py

# HTTP
pytest tests/test_http_endpoints.py

依赖项

工作室

  • fastmcp>=0.1.0 -用于简化MCP实施的FastMCP框架

SSE/HTTP

  • fastapi>=0.100.0 -现代Python web框架
  • fastapi-mcp>=0.1.0 -FastAPI MCP集成桥
  • uvicorn>=0.23.0 -FastAPI的ASGI服务器
  • python-multipart>=0.0.6 -表单数据解析
  • sse-starlette>=1.6.0 -服务器发送事件支持（隐式）

许可证

该项目在the Unlicense下发布到公共领域。有关详细信息，请参阅LICENSE文件。

未来的增强功能

  • 为双向流媒体添加WebSocket传输选项
  • 实现HTTP版本的身份验证/授权
  • 使用Pydantic模型添加请求/响应验证
  • 为这两个实现创建Docker容器
  • 添加OpenTetry仪器以实现可观察性
  • 实现工具结果缓存
  • 添加速率限制和配额管理
  • 支持批处理工具调用
  • 添加健康检查和指标端点

参考文献

存储库结构：

MCP/
├── LICENSE              # Public domain license
├── README.md           # This file
├── STDIO/              # Local process communication
│   ├── fastmcp_calculator.py
│   └── meraenv/        # Virtual environment
└── SSE/                # HTTP/SSE network communication
    └── fastapi-mcp_calculator.py

有关问题、议题或贡献，请参阅存储库文档或创建议题。

相关 MCP