AI导哈

Claude MCP流式演示

一个最小化实现Claude API流式传输与MCP集成的演示项目，用于无缝对接Vercel AI SDK。

  • by mch-ats
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 mch-ats/claude-mcp-simple-demo-

使用说明

Claude MCP流媒体演示

Claude API流与MCP（模型上下文协议）集成的最小实现，输出AI SDK流格式，用于无缝Vercel AI SDK集成。

为什么存在此演示

这表明 正确途径 使用Vercel AI SDK的MCP工具代理Claude的流媒体API。主要特点：

  1. AI SDK流格式 -将Anthropic SSE转换为AI SDK数据流协议
  2. 工具可见性 -MCP工具调用和结果在UI中可见
  3. 单转换点 -Python处理所有格式转换
  4. 工作MCP集成 -使用来自mcpservers.org的Fetch服务器

快速开始

1.安装依赖项

cd claude-mcp-simple-demo
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
pip install -r requirements.txt

2.配置环境

cp .env.example .env
# Edit .env with your actual values

所需的环境变量：

  • ANTHROPIC_API_KEY -您的Anthropic API密钥

3.运行服务器

python main.py
# Or with uvicorn:
uvicorn main:app --reload --host 0.0.0.0 --port 8000

API终点

POST /chat -流媒体

服务器发送事件（SSE）流与Vercel AI SDK兼容。

curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "Tell me a story"}' \
  --no-buffer

AI SDK示例：

import { streamText } from 'ai';

const result = await streamText({
  model: 'claude-haiku-4-5', // Model name doesn't matter, server uses env
  messages: [{ role: 'user', content: 'Fetch content from https://example.com' }],
  streamProtocol: 'data', // AI SDK data stream protocol
  experimental_telemetry: {
    isEnabled: true,
  },
});

// Stream will show:
// "[Using tool: fetch from fetch]"
// "[Tool input: {...}]"
// "I've successfully fetched the content..."

for await (const chunk of result.textStream) {
  process.stdout.write(chunk);
}

关键区别：此服务器输出AI SDK的数据流格式(0:"text"\n)，不是原始的Anthropic SSE。工具调用被转换为可见的文本标记。

流格式

服务器将Anthropic的SSE事件转换为AI SDK的数据流协议：

人类学SSE → AI SDK数据流

event: content_block_delta      →  0:"text chunk"
data: {"delta":{"text":"hi"}}   →  0:"hi"

event: message_stop             →  d:{}

这种方法的好处：

  • ✅ 工具调用在UI中可见（不隐藏在元数据中）
  • ✅ 适用于任何AI SDK消费者
  • ✅ 转换逻辑的单一真实来源
  • ✅ 更简单的客户端代码

输出示例：

0:"[Using tool: fetch from fetch]"
0:"[Tool input: {\"url\":\"https://example.com\"}]"
0:"I've successfully fetched the content..."
d:{}

MCP支持

此演示包括使用以下工具进行MCP（模型上下文协议）集成 获取 服务器从 mcpservers.org.

活动MCP服务器：

  • 获取: https://remote.mcpservers.org/fetch/mcp -提供web内容获取功能

试试看：

curl -N -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "fetch content from https://example.com"}' \
  --no-buffer

其他可用服务器 （参见 main.py 例如）：

  • github： https://api.githubcopilot.com/mcp/ （需要身份验证令牌）
  • Notion、Sentry、Linear、Figma等 mcpservers.org

重要：Anthropic API仅支持 基于URL的MCP服务器 (type: "url")，而不是基于stdio的服务器。Stdio传输仅在Claude Desktop中可用，而不在远程API中可用。

体系结构决策

✅ 这个演示做对了什么

  1. AI SDK流格式

    • 将Anthropic SSE转换为AI SDK数据流协议
    • 格式： 0:"text"\n 对于文本， d:{}\n 完成
    • 与Vercel AI SDK直接兼容

  2. 工具可见性

    • MCP工具调用在流中可见： [Using tool: fetch]
    • 显示的工具输入： [Tool input: {...}]
    • 响应中包含工具结果

  3. 单一转换点

    • Python服务器处理所有格式转换
    • 客户端仅消耗标准AI SDK流
    • 无需复杂的客户端解析

  4. 最新人类API

    • Claude Haiku 4.5（最快、最具成本效益的型号）
    • API版本2023-06-01（当前稳定版本）
    • MCP测试版：MCP-client-2025-04-04

  5. 工作MCP集成

    • 从mcpservers.org获取服务器
    • 测试不需要身份验证
    • 易于添加更多MCP服务器

❌ 这个演示故意跳过了什么

为了保持专注，此演示不包括：

  • 会话管理/对话历史记录
  • Redis或数据库持久化
  • 身份验证/速率限制
  • 自定义响应转换
  • 复杂错误恢复

根据需要在生产实现中添加这些功能。

使用不同提示进行测试

# Simple question (no tools)
curl -N -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "What is 2+2?"}' \
  --no-buffer

# Check available tools
curl -N -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "What tools do you have?"}' \
  --no-buffer

# Use MCP fetch tool
curl -N -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "Fetch and summarize https://example.com"}' \
  --no-buffer

扩展此演示

要从生产应用程序添加功能，请执行以下操作：

添加会话历史记录

# Before payload creation:
history = load_history_from_redis(session_id)
messages = history + [{"role": "user", "content": body.message}]

payload = {
    "messages": messages,  # Use full history
    # ... rest of config
}

# After streaming completes:
save_to_redis(session_id, messages + [{"role": "assistant", "content": full_response}])

捕获流媒体内容

def event_generator():
    accumulated = ""
    for line in response.iter_lines():
        if line:
            decoded = line.decode("utf-8")

            # Parse to accumulate (optional)
            if decoded.startswith("data:"):
                try:
                    data = json.loads(decoded[5:])
                    if "delta" in data and "text" in data["delta"]:
                        accumulated += data["delta"]["text"]
                except:
                    pass

            # Always forward original
            yield decoded + "\n"

    # Save after streaming
    save_to_history(session_id, accumulated)

关键要点

  1. 不要转换SSE事件 -按原样转发
  2. 使用适当的媒体类型 - text/event-stream 对于SSE
  3. 保持简单 -复杂性应该被选择加入
  4. 使用标准工具进行测试 -curl、EventSource、AI SDK

许可证

MIT-将此作为您自己实现的参考

克劳德mcp简单演示-

相关 MCP