使用Claude 搭建MCP服务访问demo服务器
本文目的是创建一个MCP服务http://127.0.0.1:5002/sse和一个外部demo服务器http://127.0.0.1:5001
启动claude后 会自动启动mcp服务，可以通过MCP http://127.0.0.1:5002/sse 访问外部服务http://127.0.0.1:5001。
以下是相关文件内容
.mcp.json文件
{
"mcpServers": {
"DemoTaskBridge": {
"command": "python",
"args": ["D:/deepseek/mcp_bridge_server.py"]
}
},
"enabledMcpjsonServers": [
"DemoTaskBridge"
]
}
config.json
{
"template_path": "assets/template.docx",
"default_name": "蔓热苺",
"output_dir": "./reports"
}
demo_server.py
from flask import Flask, request, jsonify
app = Flask(__name__)
# 模拟数据
_tasks = [
{"id": 1, "title": "finish weekly report", "status": "未完成"},
{"id": 2, "title": "code review", "status": "已完成"},
{"id": 3, "title": "fix Bug #1024", "status": "未完成"},
{"id": 4, "title": "study ai", "status": "进行中"},
]
_next_id = 5
@app.route("/")
def index():
return jsonify({"service": "demo", "version": "1.0", "endpoints": ["/api/tasks", "/api/echo"]})
@app.route("/api/tasks", methods=["GET", "POST"])
def tasks():
if request.method == "GET":
return jsonify(_tasks)
elif request.method == "POST":
global _next_id
data = request.get_json(force=True)
status = data.get("status", "未完成")
task = {"id": _next_id, "title": data["title"], "status": status}
_next_id += 1
_tasks.append(task)
return jsonify(task), 201
@app.route("/api/tasks/<int:task_id>", methods=["GET", "PUT", "DELETE"])
def task_detail(task_id):
task = next((t for t in _tasks if t["id"] == task_id), None)
if task is None:
return jsonify({"error": "not found"}), 404
if request.method == "GET":
return jsonify(task)
elif request.method == "PUT":
data = request.get_json(force=True)
task.update(data)
return jsonify(task)
elif request.method == "DELETE":
_tasks.remove(task)
return jsonify({"deleted": task_id})
@app.route("/api/echo", methods=["GET", "POST"])
def echo():
if request.method == "GET":
return jsonify({"args": request.args.to_dict()})
return jsonify({"method": "POST", "body": request.get_json(silent=True) or {}, "args": request.args.to_dict()})
if __name__ == "__main__":
app.run(host="127.0.0.1", port=5001, debug=False)
mcp_bridge_server.py
from fastmcp import FastMCP
import httpx
import json
import subprocess
import sys
import atexit
import time
# 自动启动 demo 服务
_demo_proc = subprocess.Popen(
[sys.executable, "demo_server.py"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
atexit.register(lambda: _demo_proc.terminate())
time.sleep(0.5) # 等待 demo 服务就绪
# 初始化 MCP 服务器
mcp = FastMCP("DemoTaskBridge")
# 配置原有服务的地址
ORIGINAL_SERVICE_URL = "http://127.0.0.1:5001"
@mcp.tool()
async def call_original_api(endpoint: str, method: str = "GET", params: dict = None, body: dict = None) -> str:
"""
调用本地 127.0.0.1:5001 服务的通用工具。
Args:
endpoint: API 路径，例如 '/api/data' 或 '/users'
method: HTTP 方法, GET, POST, PUT, DELETE 等
params: URL 查询参数 (字典)
body: 请求体数据 (字典，用于 POST/PUT)
Returns:
服务返回的 JSON 字符串或错误信息
"""
url = f"{ORIGINAL_SERVICE_URL}{endpoint}"
try:
async with httpx.AsyncClient(timeout=10.0) as client:
if method.upper() == "GET":
response = await client.get(url, params=params)
elif method.upper() == "POST":
response = await client.post(url, json=body, params=params)
elif method.upper() == "PUT":
response = await client.put(url, json=body, params=params)
elif method.upper() == "DELETE":
response = await client.delete(url, params=params)
else:
return json.dumps({"error": f"Unsupported method: {method}"})
# 尝试解析 JSON，如果失败则返回文本
try:
result = response.json()
return json.dumps(result, ensure_ascii=False)
except json.JSONDecodeError:
return response.text
except httpx.ConnectError:
return json.dumps({"error": "无法连接到本地服务 127.0.0.1:5001,请确保服务已启动"})
except Exception as e:
return json.dumps({"error": str(e)})
if __name__ == "__main__":
# 支持 --sse 参数用于调试，默认 stdio（Claude Code 自动拉起）
if len(sys.argv) > 1 and sys.argv[1] == "--sse":
mcp.run(transport="sse", host="127.0.0.1", port=5002)
else:
mcp.run(transport="stdio")
skills.py
import os
import json
import pandas as pd
from datetime import datetime, timedelta
from docx import Document
from docx.shared import Inches, Pt
from docx.enum.text import WD_ALIGN_PARAGRAPH
from docx.oxml.ns import qn
class WeeklyReportGenerator:
def __init__(self, config_path='config.json'):
"""初始化周报生成器"""
with open(config_path, 'r', encoding='utf-8') as f:
self.config = json.load(f)
self.report_data = {
"name": "",
"week_range": "",
"completed_tasks": [],
"ongoing_tasks": [],
"issues": [],
"next_plan": []
}
def load_data_from_excel(self, file_path):
"""从Excel文件加载工作记录"""
try:
df = pd.read_excel(file_path, engine='openpyxl')
# 假设Excel列名为: 日期, 事项, 状态(已完成/进行中), 备注
for _, row in df.iterrows():
task_info = {
"date": str(row.get('日期', '')),
"content": str(row.get('事项', '')),
"status": str(row.get('状态', '')),
"remark": str(row.get('备注', ''))
}
if '完成' in task_info['status']:
self.report_data['completed_tasks'].append(task_info)
else:
self.report_data['ongoing_tasks'].append(task_info)
print(f"成功从 {file_path} 加载 {len(df)} 条记录")
except Exception as e:
print(f"读取Excel失败: {e}")
def load_data_from_log(self, log_file_path):
"""从文本日志加载工作记录 (简易解析)"""
try:
with open(log_file_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
current_date = ""
for line in lines:
line = line.strip()
if not line:
continue
# 简单判断日期行 (如 12.21 或 2026-07-21)
if '.' in line and len(line) < 15 and line.isdigit():
current_date = line
continue
# 假设非日期行均为工作内容，简单归类为已完成
if current_date:
self.report_data['completed_tasks'].append({
"date": current_date,
"content": line,
"status": "已完成",
"remark": ""
})
print(f"成功从日志 {log_file_path} 解析记录")
except Exception as e:
print(f"读取日志失败: {e}")
def generate_word_report(self, output_filename=None):
"""生成Word格式周报"""
if not output_filename:
output_filename = f"WeeklyReport_{datetime.now().strftime('%Y%m%d')}.docx"
doc = Document()
# 设置中文字体支持
style = doc.styles['Normal']
style.font.name = u'微软雅黑'
style._element.rPr.rFonts.set(qn('w:eastAsia'), u'微软雅黑')
# 标题
title = doc.add_heading('工作周报', 0)
title.alignment = WD_ALIGN_PARAGRAPH.CENTER
# 基本信息
p_info = doc.add_paragraph()
p_info.add_run(f"姓名: {self.report_data['name'] or '未填写'}\t\t")
p_info.add_run(f"周期: {self.report_data['week_range'] or '本周'}")
# 1. 本周完成工作
doc.add_heading('一、本周完成工作', level=1)
if self.report_data['completed_tasks']:
for task in self.report_data['completed_tasks']:
p = doc.add_paragraph(style='List Bullet')
p.add_run(f"[{task['date']}] {task['content']}")
if task['remark']:
p.add_run(f" (备注: {task['remark']})")
else:
doc.add_paragraph('无记录')
# 2. 进行中工作
doc.add_heading('二、进行中工作', level=1)
if self.report_data['ongoing_tasks']:
for task in self.report_data['ongoing_tasks']:
p = doc.add_paragraph(style='List Bullet')
p.add_run(f"[{task['date']}] {task['content']} - 状态: {task['status']}")
else:
doc.add_paragraph('无记录')
# 3. 问题与风险
doc.add_heading('三、问题与风险', level=1)
if self.report_data['issues']:
for issue in self.report_data['issues']:
doc.add_paragraph(issue, style='List Number')
else:
doc.add_paragraph('暂无重大风险')
# 4. 下周计划
doc.add_heading('四、下周计划', level=1)
if self.report_data['next_plan']:
for plan in self.report_data['next_plan']:
doc.add_paragraph(plan, style='List Number')
else:
doc.add_paragraph('按常规项目进度推进')
# 保存文档
doc.save(output_filename)
print(f"周报已生成: {output_filename}")
return output_filename
def main():
"""主入口函数"""
# 初始化生成器
generator = WeeklyReportGenerator()
# 模拟配置用户信息
generator.report_data['name'] = "AI助手"
generator.report_data['week_range'] = "2026年第29周 (07.13-07.19)"
# 示例1: 从Excel加载 (如果存在 test_data.xlsx)
if os.path.exists('test_data.xlsx'):
generator.load_data_from_excel('test_data.xlsx')
else:
# 示例2: 从日志加载 (如果存在 work_log.txt)
if os.path.exists('work_log.txt'):
generator.load_data_from_log('work_log.txt')
else:
# 如果没有数据源，使用模拟数据演示
print("未找到数据源，使用模拟数据...")
generator.report_data['completed_tasks'] = [
{"date": "07.15", "content": "完成周报Skill模块开发", "status": "已完成", "remark": "包含Excel解析功能"},
{"date": "07.16", "content": "优化Prompt提示词结构", "status": "已完成", "remark": ""}
]
generator.report_data['ongoing_tasks'] = [
{"date": "07.17", "content": "集成大模型API接口", "status": "进行中", "remark": "等待Key审批"}
]
generator.report_data['next_plan'] = [
"完成单元测试覆盖",
"部署到生产环境"
]
# 生成报告
generator.generate_word_report()
if __name__ == '__main__':
main()
start_service.py
#!/usr/bin/env python
"""服务管理脚本：支持手动启动/停止，也可由 Claude Code 自动拉起"""
import subprocess
import sys
import urllib.request
import os
if sys.platform == "win32":
sys.stdout.reconfigure(encoding="utf-8")
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
SERVICES = {
"1": {
"name": "Demo 服务",
"port": 5001,
"script": "demo_server.py",
"desc": "Flask 模拟 API，可直接测试",
},
"2": {
"name": "MCP Bridge (SSE)",
"port": 5002,
"script": "mcp_bridge_server.py",
"args": ["--sse"],
"desc": "FastMCP 桥接，SSE 模式调试用",
},
}
def check_port(port):
"""检查端口是否在监听"""
try:
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(2)
s.connect(("127.0.0.1", port))
s.close()
return True
except Exception:
return False
def find_pid(port):
"""查找占用端口的 PID"""
try:
result = subprocess.run(
["netstat", "-ano"], capture_output=True, text=True
)
for line in result.stdout.splitlines():
if f":{port}" in line and "LISTENING" in line:
return line.strip().split()[-1]
except Exception:
pass
return None
def kill_service(port):
"""停止指定端口的服务"""
pid = find_pid(port)
if pid:
subprocess.run(["cmd", "/c", f"taskkill /PID {pid} /F"],
capture_output=True)
return True
return False
def show_status():
print("\n" + "=" * 50)
print(" 服务状态")
print("=" * 50)
for key, svc in SERVICES.items():
running = check_port(svc["port"])
status = "运行中" if running else "未运行"
pid = find_pid(svc["port"])
pid_str = f"(PID: {pid})" if pid else ""
print(f" [{key}] {svc['name']:<20} :{svc['port']:<6} {status} {pid_str}")
print("=" * 50)
print(" 启动方式：")
print(" A. Claude Code 自动拉起 (默认) — 无需任何操作")
print(" B. 脚本手动启动 — 使用下方菜单")
print("=" * 50)
def show_menu():
print("\n [1] 启动 Demo 服务")
print(" [2] 启动 MCP Bridge (SSE 调试模式)")
print(" [3] 停止 Demo 服务")
print(" [4] 停止 MCP Bridge")
print(" [s] 刷新状态")
print(" [q] 退出")
def start_service(key):
svc = SERVICES[key]
if check_port(svc["port"]):
print(f" {svc['name']} 已在运行中")
return None
print(f" 启动 {svc['name']} (:{svc['port']}) ...")
args = [sys.executable, os.path.join(SCRIPT_DIR, svc["script"])]
if "args" in svc:
args.extend(svc["args"])
return subprocess.Popen(args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
def stop_service(key):
svc = SERVICES[key]
if not check_port(svc["port"]):
print(f" {svc['name']} 未在运行")
return
print(f" 停止 {svc['name']} (:{svc['port']}) ...")
if kill_service(svc["port"]):
print(" 已停止")
else:
print(" 停止失败")
def main():
procs = {}
while True:
show_status()
show_menu()
try:
choice = input("\n请选择: ").strip()
except (EOFError, KeyboardInterrupt):
print()
break
if choice == "q":
break
elif choice == "s":
continue
elif choice == "1":
p = start_service("1")
if p:
procs["1"] = p
elif choice == "2":
p = start_service("2")
if p:
procs["2"] = p
elif choice == "3":
stop_service("1")
elif choice == "4":
stop_service("2")
else:
print(" 无效选择")
# 清理
for p in procs.values():
p.terminate()
if __name__ == "__main__":
main()
settings.local.json
{
"permissions": {
"allow": [
"Bash(python -c \"import akshare as ak; help\\(ak.stock_board_concept_cons_em\\)\")",
"Bash(python -c ' *)",
"Bash(python -c \"import py_compile; py_compile.compile\\('macd_golden_cross_hot.py', doraise=True\\); print\\('OK - no syntax errors'\\)\")",
"Bash(python macd_golden_cross_hot.py)",
"Bash(python mcp_bridge_server.py)",
"Bash(python *)",
"Bash(curl -s http://127.0.0.1:5001/)",
"Bash(curl -s http://127.0.0.1:5001/api/tasks)",
"Bash(curl -s -X POST http://127.0.0.1:5001/api/tasks -H \"Content-Type: application/json\" -d \"{\\\\\"title\\\\\":\\\\\"新任务\\\\\"}\")",
"Bash(curl -s \"http://127.0.0.1:5001/api/echo?name=test\")",
"Bash(curl -s -X POST http://127.0.0.1:5001/api/echo -H \"Content-Type: application/json\" -d \"{\\\\\"msg\\\\\":\\\\\"hello\\\\\"}\")",
"Bash(netstat -ano)",
"Bash(taskkill *)",
"Bash(curl -s http://127.0.0.1:5002/sse)",
"Bash(curl -s http://127.0.0.1:5002/)",
"Bash(pip install *)",
"Bash(curl -s -o /dev/null -w \"%{http_code}\" http://127.0.0.1:5001/)",
"Bash(curl *)",
"Bash(cmd *)"
]
}
}
mcp_bridge_doc.md
# DemoTaskBridge MCP 服务文档
> MCP 桥接服务，将 Demo Server (5001) 的 REST API 暴露为 MCP 工具，供 Claude Code 等 AI 客户端调用。
## 架构
```
Claude Code ──(stdio/SSE)──> DemoTaskBridge (5002) ──(HTTP)──> Demo Server (5001)
MCP 客户端 FastMCP 桥接 Flask REST API
```
## 两种启动方式
### 方式 A：Claude Code 自动拉起（推荐）
`.mcp.json` 配置 stdio 模式，Claude Code 启动时自动执行，无需手动干预。
```json
{
"mcpServers": {
"DemoTaskBridge": {
"command": "python",
"args": ["D:/deepseek/mcp_bridge_server.py"]
}
}
}
```
### 方式 B：手动 SSE 模式（调试用）
```bash
python mcp_bridge_server.py --sse
# 服务监听 http://127.0.0.1:5002/sse
# 或使用脚本
python start_service.py
```
---
## MCP 工具
### call_original_api
调用 Demo Server 的通用 HTTP 工具。
| 参数 | 类型 | 必填 | 默认值 | 说明 |
| -------- | ------ | ---- | ------ | --------------------------- |
| endpoint | string | 是 | - | API 路径，如 `/api/tasks` |
| method | string | 否 | GET | HTTP 方法：GET/POST/PUT/DELETE |
| params | dict | 否 | - | URL 查询参数 |
| body | dict | 否 | - | 请求体（POST/PUT） |
**返回**：Demo Server 的 JSON 响应字符串。
---
## 使用示例
### 在 Claude Code 中（自然语言）
```
查一下有哪些任务
创建一个任务：标题=学习Python，状态=进行中
把 id=3 的任务状态改为已完成
删除 id=1 的任务
```
### MCP 协议调用（JSON-RPC over SSE）
**获取任务列表**
```json
// → tools/call
{
"name": "call_original_api",
"arguments": {
"endpoint": "/api/tasks"
}
}
// ← 返回
[
{"id": 1, "title": "finish weekly report", "status": "未完成"},
{"id": 2, "title": "code review", "status": "已完成"}
]
```
**创建任务**
```json
// → tools/call
{
"name": "call_original_api",
"arguments": {
"endpoint": "/api/tasks",
"method": "POST",
"body": {"title": "学习Python", "status": "进行中"}
}
}
// ← 返回 (201)
{"id": 5, "title": "学习Python", "status": "进行中"}
```
**更新任务**
```json
// → tools/call
{
"name": "call_original_api",
"arguments": {
"endpoint": "/api/tasks/3",
"method": "PUT",
"body": {"status": "已完成"}
}
}
// ← 返回
{"id": 3, "title": "fix Bug #1024", "status": "已完成"}
```
**删除任务**
```json
// → tools/call
{
"name": "call_original_api",
"arguments": {
"endpoint": "/api/tasks/1",
"method": "DELETE"
}
}
// ← 返回
{"deleted": 1}
```
**Echo 测试**
```json
// → tools/call
{
"name": "call_original_api",
"arguments": {
"endpoint": "/api/echo",
"params": {"name": "test"}
}
}
// ← 返回
{"args": {"name": "test"}}
```
---
## MCP 协议握手流程
若需手动调试 MCP 协议，完整流程如下：
```
1. GET /sse → 获取 session_id
2. POST /messages/?session_id= → initialize（握手）
3. POST /messages/?session_id= → notifications/initialized
4. POST /messages/?session_id= → tools/call（调用工具）
```
Python 示例：
```python
import httpx, asyncio, json
async def call_mcp_tool(endpoint, method="GET", params=None, body=None):
async with httpx.AsyncClient() as client:
async with client.stream("GET", "http://127.0.0.1:5002/sse") as sse:
lines = sse.aiter_lines()
session_id = None
async for line in lines:
if "session_id=" in line:
session_id = line.split("session_id=")[1]
break
async def rpc(method_name, p=None):
await client.post(
f"http://127.0.0.1:5002/messages/?session_id={session_id}",
json={"jsonrpc": "2.0", "id": 1, "method": method_name, "params": p or {}}
)
async for line in lines:
if line.startswith("data: "):
return json.loads(line[6:])
# 握手
await rpc("initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "cli", "version": "1.0"}
})
await client.post(
f"http://127.0.0.1:5002/messages/?session_id={session_id}",
json={"jsonrpc": "2.0", "method": "notifications/initialized"}
)
# 调用工具
result = await rpc("tools/call", {
"name": "call_original_api",
"arguments": {"endpoint": endpoint, "method": method, "params": params, "body": body}
})
return json.loads(result["result"]["content"][0]["text"])
```
---
## 错误处理
| 错误 | 说明 |
| ------------------------------ | ------------------------ |
| `无法连接到本地服务` | Demo Server (5001) 未启动 |
| `{"error": "not found"}` | 任务 ID 不存在 (404) |
| `Unsupported method: xxx` | HTTP 方法不支持 |
---
## 文件清单
| 文件 | 说明 |
| ------------------------- | ------------------------ |
| `mcp_bridge_server.py` | MCP 桥接服务主程序 |
| `demo_server.py` | Demo REST API 服务 |
| `.mcp.json` | Claude Code MCP 配置 |
| `start_service.py` | 服务管理脚本 |
| `demo_api.md` | Demo API 文档 |
demo_api.md
# Demo Server API 文档
> 服务地址：`http://127.0.0.1:5001`
## 概览
| 方法 | 路径 | 说明 |
| ------ | ----------------- | ------------ |
| GET | `/` | 服务信息 |
| GET | `/api/tasks` | 获取任务列表 |
| POST | `/api/tasks` | 创建任务 |
| GET | `/api/tasks/<id>` | 获取单个任务 |
| PUT | `/api/tasks/<id>` | 更新任务 |
| DELETE | `/api/tasks/<id>` | 删除任务 |
| GET | `/api/echo` | Echo 测试 |
| POST | `/api/echo` | Echo 测试 |
---
## 任务字段
| 字段 | 类型 | 说明 |
| ------ | ------ | ------------------------------- |
| id | int | 任务 ID（自动分配） |
| title | string | 任务标题（必填） |
| status | string | 状态：`未完成` `进行中` `已完成` |
---
## 接口详情
### GET / — 服务信息
**请求**
```bash
curl -s http://127.0.0.1:5001/
```
**响应** `200`
```json
{
"service": "demo",
"version": "1.0",
"endpoints": ["/api/tasks", "/api/echo"]
}
```
---
### GET /api/tasks — 获取任务列表
**请求**
```bash
curl -s http://127.0.0.1:5001/api/tasks
```
**响应** `200`
```json
[
{"id": 1, "title": "finish weekly report", "status": "未完成"},
{"id": 2, "title": "code review", "status": "已完成"},
{"id": 3, "title": "fix Bug #1024", "status": "未完成"}
]
```
---
### POST /api/tasks — 创建任务
**请求**
```bash
curl -s -X POST http://127.0.0.1:5001/api/tasks \
-H "Content-Type: application/json" \
-d '{"title": "新任务"}'
# 指定状态
curl -s -X POST http://127.0.0.1:5001/api/tasks \
-H "Content-Type: application/json" \
-d '{"title": "新任务", "status": "进行中"}'
```
| 参数 | 类型 | 必填 | 默认值 | 说明 |
| ------ | ------ | ---- | -------- | ------------------------------- |
| title | string | 是 | - | 任务标题 |
| status | string | 否 | `未完成` | `未完成` `进行中` `已完成` |
**响应** `201`
```json
{"id": 5, "title": "新任务", "status": "未完成"}
```
---
### GET /api/tasks/<id> — 获取单个任务
**请求**
```bash
curl -s http://127.0.0.1:5001/api/tasks/1
```
**响应** `200`
```json
{"id": 1, "title": "finish weekly report", "status": "未完成"}
```
**404**
```json
{"error": "not found"}
```
---
### PUT /api/tasks/<id> — 更新任务
**请求**
```bash
curl -s -X PUT http://127.0.0.1:5001/api/tasks/1 \
-H "Content-Type: application/json" \
-d '{"status": "已完成"}'
```
**响应** `200`
```json
{"id": 1, "title": "finish weekly report", "status": "已完成"}
```
---
### DELETE /api/tasks/<id> — 删除任务
**请求**
```bash
curl -s -X DELETE http://127.0.0.1:5001/api/tasks/1
```
**响应** `200`
```json
{"deleted": 1}
```
---
### GET /api/echo — Echo 测试
**请求**
```bash
curl -s "http://127.0.0.1:5001/api/echo?name=test&page=1"
```
**响应** `200`
```json
{"args": {"name": "test", "page": "1"}}
```
---
### POST /api/echo — Echo 测试
**请求**
```bash
curl -s -X POST http://127.0.0.1:5001/api/echo \
-H "Content-Type: application/json" \
-d '{"msg": "hello"}'
```
**响应** `200`
```json
{
"method": "POST",
"body": {"msg": "hello"},
"args": {}
}
```
汇聚全球AI编程工具，助力开发者即刻编程。
更多推荐
6
0- 0
已为社区贡献2条内容
所有评论(0)