使用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": {}

}

```