sse mcp
This commit is contained in:
@@ -6,25 +6,42 @@
|
||||
- list_docx_images:列出 DOCX 中的图片信息
|
||||
- edit_docx: 进行文本替换 / 关键字上色 / 图片替换
|
||||
|
||||
支持两种传输方式:
|
||||
- stdio(默认,本地使用)
|
||||
- sse(远程调用,通过 HTTP SSE 协议)
|
||||
|
||||
用法:
|
||||
# 本地 stdio 模式
|
||||
python mcp_docx_server.py
|
||||
|
||||
# SSE 远程模式(默认 0.0.0.0:8080)
|
||||
python mcp_docx_server.py --transport sse
|
||||
python mcp_docx_server.py --transport sse --host 0.0.0.0 --port 8080
|
||||
|
||||
# 客户端连接地址:
|
||||
# SSE 端点: http://<host>:<port>/sse
|
||||
# 消息端点: http://<host>:<port>/messages/
|
||||
|
||||
注意:底层仍然完全复用 mcp_docx.py 中的逻辑,只是通过 MCP SDK 对外提供。
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import os
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from mcp.server.fastmcp import FastMCPServer
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
|
||||
from mcp_docx import get_images_info, process, _parse_span_replacement
|
||||
|
||||
|
||||
server = FastMCPServer(
|
||||
mcp = FastMCP(
|
||||
"docx-editor",
|
||||
version="0.1.0",
|
||||
description="DOCX 文本和图片编辑工具(基于 mcp_docx.py 封装)",
|
||||
)
|
||||
|
||||
|
||||
@server.tool()
|
||||
@mcp.tool()
|
||||
async def list_docx_images(docx_path: str) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
列出指定 DOCX 文件中的所有图片信息。
|
||||
@@ -36,7 +53,6 @@ async def list_docx_images(docx_path: str) -> List[Dict[str, Any]]:
|
||||
- 图片信息列表,每一项包含:
|
||||
- index: 图片在文档中的顺序(从 1 开始)
|
||||
- media_file: DOCX 内部的资源路径
|
||||
- abs_path: 解包后的绝对路径(仅用于调试)
|
||||
- ext: 图片扩展名
|
||||
- docpr_name: Word 内部的图片名称
|
||||
- width_cm / height_cm: 近似尺寸(厘米),可能为 None
|
||||
@@ -45,13 +61,13 @@ async def list_docx_images(docx_path: str) -> List[Dict[str, Any]]:
|
||||
raise FileNotFoundError(f"DOCX 文件不存在: {docx_path}")
|
||||
|
||||
imgs = get_images_info(docx_path)
|
||||
# 为了避免泄露容器内部路径,可选择屏蔽 abs_path 字段
|
||||
# 为了避免泄露容器内部路径,屏蔽 abs_path 字段
|
||||
for img in imgs:
|
||||
img.pop("abs_path", None)
|
||||
return imgs
|
||||
|
||||
|
||||
@server.tool()
|
||||
@mcp.tool()
|
||||
async def edit_docx(
|
||||
input_docx: str,
|
||||
output_docx: str,
|
||||
@@ -132,5 +148,32 @@ async def edit_docx(
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# 通过 stdio 运行 MCP 服务器
|
||||
server.run()
|
||||
parser = argparse.ArgumentParser(description="DOCX MCP 服务器")
|
||||
parser.add_argument(
|
||||
"--transport",
|
||||
choices=["stdio", "sse"],
|
||||
default="stdio",
|
||||
help="传输方式:stdio(本地)或 sse(远程 SSE)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--host",
|
||||
default="0.0.0.0",
|
||||
help="SSE 模式监听地址(默认 0.0.0.0)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--port",
|
||||
type=int,
|
||||
default=8080,
|
||||
help="SSE 模式监听端口(默认 8080)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.transport == "sse":
|
||||
# SSE 远程模式:通过 HTTP 暴露 MCP 服务
|
||||
mcp.settings.host = args.host
|
||||
mcp.settings.port = args.port
|
||||
print(f"🚀 MCP SSE 服务器启动中 → http://{args.host}:{args.port}/sse")
|
||||
mcp.run(transport="sse")
|
||||
else:
|
||||
# 本地 stdio 模式
|
||||
mcp.run(transport="stdio")
|
||||
|
||||
Reference in New Issue
Block a user