本文目的是创建一个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.pyfrom 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(forceTrue) 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(forceTrue) 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(silentTrue) or {}, args: request.args.to_dict()}) if __name__ __main__: app.run(host127.0.0.1, port5001, debugFalse)mcp_bridge_server.pyfrom 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], stdoutsubprocess.DEVNULL, stderrsubprocess.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(timeout10.0) as client: if method.upper() GET: response await client.get(url, paramsparams) elif method.upper() POST: response await client.post(url, jsonbody, paramsparams) elif method.upper() PUT: response await client.put(url, jsonbody, paramsparams) elif method.upper() DELETE: response await client.delete(url, paramsparams) else: return json.dumps({error: fUnsupported method: {method}}) # 尝试解析 JSON如果失败则返回文本 try: result response.json() return json.dumps(result, ensure_asciiFalse) 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 参数用于调试默认 stdioClaude Code 自动拉起 if len(sys.argv) 1 and sys.argv[1] --sse: mcp.run(transportsse, host127.0.0.1, port5002) else: mcp.run(transportstdio)skills.pyimport 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_pathconfig.json): 初始化周报生成器 with open(config_path, r, encodingutf-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, engineopenpyxl) # 假设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, encodingutf-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_filenameNone): 生成Word格式周报 if not output_filename: output_filename fWeeklyReport_{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(一、本周完成工作, level1) if self.report_data[completed_tasks]: for task in self.report_data[completed_tasks]: p doc.add_paragraph(styleList 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(二、进行中工作, level1) if self.report_data[ongoing_tasks]: for task in self.report_data[ongoing_tasks]: p doc.add_paragraph(styleList Bullet) p.add_run(f[{task[date]}] {task[content]} - 状态: {task[status]}) else: doc.add_paragraph(无记录) # 3. 问题与风险 doc.add_heading(三、问题与风险, level1) if self.report_data[issues]: for issue in self.report_data[issues]: doc.add_paragraph(issue, styleList Number) else: doc.add_paragraph(暂无重大风险) # 4. 下周计划 doc.add_heading(四、下周计划, level1) if self.report_data[next_plan]: for plan in self.report_data[next_plan]: doc.add_paragraph(plan, styleList 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(encodingutf-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_outputTrue, textTrue ) 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, ftaskkill /PID {pid} /F], capture_outputTrue) 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, stdoutsubprocess.DEVNULL, stderrsubprocess.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, doraiseTrue\\); 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?nametest\),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## 两种启动方式### 方式 AClaude Code 自动拉起推荐.mcp.json 配置 stdio 模式Claude Code 启动时自动执行无需手动干预。json{mcpServers: {DemoTaskBridge: {command: python,args: [D:/deepseek/mcp_bridge_server.py]}}}### 方式 B手动 SSE 模式调试用bashpython 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状态进行中把 id3 的任务状态改为已完成删除 id1 的任务### 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_id2. POST /messages/?session_id → initialize握手3. POST /messages/?session_id → notifications/initialized4. POST /messages/?session_id → tools/call调用工具Python 示例pythonimport httpx, asyncio, jsonasync def call_mcp_tool(endpoint, methodGET, paramsNone, bodyNone):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 Noneasync for line in lines:if session_id in line:session_id line.split(session_id)[1]breakasync def rpc(method_name, pNone):await client.post(fhttp://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(fhttp://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 / — 服务信息**请求**bashcurl -s http://127.0.0.1:5001/**响应** 200json{service: demo,version: 1.0,endpoints: [/api/tasks, /api/echo]}---### GET /api/tasks — 获取任务列表**请求**bashcurl -s http://127.0.0.1:5001/api/tasks**响应** 200json[{id: 1, title: finish weekly report, status: 未完成},{id: 2, title: code review, status: 已完成},{id: 3, title: fix Bug #1024, status: 未完成}]---### POST /api/tasks — 创建任务**请求**bashcurl -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 | 否 | 未完成 | 未完成 进行中 已完成 |**响应** 201json{id: 5, title: 新任务, status: 未完成}---### GET /api/tasks/id — 获取单个任务**请求**bashcurl -s http://127.0.0.1:5001/api/tasks/1**响应** 200json{id: 1, title: finish weekly report, status: 未完成}**404**json{error: not found}---### PUT /api/tasks/id — 更新任务**请求**bashcurl -s -X PUT http://127.0.0.1:5001/api/tasks/1 \-H Content-Type: application/json \-d {status: 已完成}**响应** 200json{id: 1, title: finish weekly report, status: 已完成}---### DELETE /api/tasks/id — 删除任务**请求**bashcurl -s -X DELETE http://127.0.0.1:5001/api/tasks/1**响应** 200json{deleted: 1}---### GET /api/echo — Echo 测试**请求**bashcurl -s http://127.0.0.1:5001/api/echo?nametestpage1**响应** 200json{args: {name: test, page: 1}}---### POST /api/echo — Echo 测试**请求**bashcurl -s -X POST http://127.0.0.1:5001/api/echo \-H Content-Type: application/json \-d {msg: hello}**响应** 200json{method: POST,body: {msg: hello},args: {}}