#!/usr/bin/env python3 """ 基于 mcp_docx.py 封装的 MCP 服务器。 暴露两个主要工具： - list_docx_images：列出 DOCX 中的图片信息 - edit_docx: 进行文本替换 / 关键字上色 / 图片替换 当前推荐的传输方式： - stdio（本地调试） - streamable-http（远程 HTTP，路径固定为 /mcp，推荐） 用法： # 本地 stdio 模式（默认） python mcp_docx_server.py --transport stdio # HTTP 远程模式（推荐，默认 0.0.0.0:8080，对外暴露 /mcp） python mcp_docx_server.py --transport http python mcp_docx_server.py --transport http --host 0.0.0.0 --port 8080 # 客户端连接地址（http 模式）： # MCP 端点: http://:/mcp 注意：底层仍然完全复用 mcp_docx.py 中的逻辑，只是通过 MCP SDK 对外提供。 """ import argparse import os import tempfile import urllib.parse from typing import Any, Dict, List, Optional import requests from mcp.server.fastmcp import FastMCP from mcp.server.transport_security import TransportSecuritySettings from mcp_docx import get_images_info, process, _parse_span_replacement _disable_dns_rebinding = os.getenv("MCP_DISABLE_HOST_CHECK") == "1" if _disable_dns_rebinding: # 参考 python-sdk 官方文档：关闭 DNS rebinding 防护（适合本地或已由外层网关做安全控制的环境） # https://github.com/modelcontextprotocol/python-sdk/issues/1798 transport_security = TransportSecuritySettings( enable_dns_rebinding_protection=False, ) else: # 默认：开启 DNS rebinding 防护，但允许本机访问 # 如需通过网关 / 域名访问，可在这里追加 allowed_hosts / allowed_origins transport_security = TransportSecuritySettings( enable_dns_rebinding_protection=True, allowed_hosts=["localhost:*", "127.0.0.1:*","149.88.66.186:*"], allowed_origins=["http://localhost:*", "http://127.0.0.1:*","http://149.88.66.186:*"], ) mcp = FastMCP( "docx-editor", transport_security=transport_security, ) def _is_url(path: str) -> bool: """简单判断一个字符串是否为 HTTP/HTTPS URL。""" return path.startswith("http://") or path.startswith("https://") def _download_to_temp(url: str, suffix: str = ".tmp") -> str: """ 将远程 URL 下载到临时文件，返回本地临时路径。 调用方负责在使用完毕后删除该文件。 """ resp = requests.get(url, stream=True, timeout=30) resp.raise_for_status() fd, tmp_path = tempfile.mkstemp(suffix=suffix) try: with os.fdopen(fd, "wb") as f: for chunk in resp.iter_content(chunk_size=8192): if chunk: f.write(chunk) except Exception: # 出错时清理临时文件 try: os.remove(tmp_path) except OSError: pass raise return tmp_path def _build_output_url(abs_output_path: str) -> Optional[str]: """ 根据环境变量 MCP_OUTPUT_BASE_URL 构造输出文件的 URL。 约定： - MCP_OUTPUT_BASE_URL 形如: http://host:port/files/ - 最终 URL = MCP_OUTPUT_BASE_URL.rstrip('/') + '/' + 文件名 """ base = os.getenv("MCP_OUTPUT_BASE_URL") if not base: return None filename = os.path.basename(abs_output_path) return base.rstrip("/") + "/" + filename @mcp.tool() async def list_docx_images(docx_url: str) -> List[Dict[str, Any]]: """ 列出指定 DOCX 文件中的所有图片信息。 参数: - docx_url: 文件的HTTP/HTTPS URL。 返回: - 图片信息列表，每一项包含: - index: 图片在文档中的顺序（从 1 开始） - media_file: DOCX 内部的资源路径 - ext: 图片扩展名 - docpr_name: Word 内部的图片名称 - width_cm / height_cm: 近似尺寸（厘米），可能为 None """ imgs = get_images_info(_download_to_temp(docx_url, suffix=".docx")) return imgs @mcp.tool() async def edit_docx(input_docx_url: str, output_docx_url: str, replacements: Optional[List[Dict[str, str]]] = None, image_replacements: Optional[List[Dict[str, Any]]] = None) -> Dict[str, Any]: """ 使用原始 mcp_docx 逻辑对 DOCX 文件进行编辑。 支持： - 纯文本替换 - 通过 关键字 语法设置关键字颜色 - 替换指定序号的图片 参数: - input_docx_url: 输入 DOCX 文件HTTP/HTTPS URL - output_docx_url: 输出 DOCX 文件HTTP/HTTPS URL - replacements: 文本替换规则列表，例如: [ {\"old\": \"旧标题\", \"new\": \"新标题\"}, {\"old\": \"原文\", \"new\": \"新文\"} ] - image_replacements: 图片替换规则列表，例如: [ {\"index\": 1, \"file\": \"new_chart.png\"}, {\"index\": 2, \"file\": \"new_photo.jpg\"} ] 其中 file 字段同样可以是本地路径或 HTTP/HTTPS URL。 返回: - { \"output_path\": 生成的 DOCX 绝对路径, \"output_url\": 如果配置了 MCP_OUTPUT_BASE_URL，则为可访问该文件的 URL，否则为 null } """ tmp_input: Optional[str] = None tmp_images: List[str] = [] print(f"edit_docx: input_docx_url: {input_docx_url}, output_docx_url: {output_docx_url}, replacements: {replacements}, image_replacements: {image_replacements}") try: local_input = input_docx_url if _is_url(input_docx_url): parsed = urllib.parse.urlparse(input_docx_url) ext = os.path.splitext(parsed.path)[1] or ".docx" tmp_input = _download_to_temp(input_docx_url, suffix=ext) local_input = tmp_input if not os.path.exists(local_input): raise FileNotFoundError(f"输入 DOCX 文件不存在: {input_docx_url}") if replacements is None: replacements = [] if image_replacements is None: image_replacements = [] # 解析文本替换与颜色关键字（复用 CLI 逻辑） rep_pairs = [] color_keywords = [] for item in replacements: old = item.get("old") new_raw = item.get("new") if not old: continue if new_raw is None: new_raw = "" new_plain, spans = _parse_span_replacement(new_raw) rep_pairs.append((old, new_plain)) color_keywords.extend(spans) # 处理图片替换参数（支持本地路径或 URL） img_pairs = [] for item in image_replacements: try: idx = int(item.get("index")) except (TypeError, ValueError): continue path = item.get("file") if not path: continue local_img = path if _is_url(path): parsed = urllib.parse.urlparse(path) ext = os.path.splitext(parsed.path)[1] or "" suffix = ext if ext else ".img" tmp_img = _download_to_temp(path, suffix=suffix) tmp_images.append(tmp_img) local_img = tmp_img if not os.path.exists(local_img): raise FileNotFoundError(f"图片文件不存在: {path}") img_pairs.append((idx, local_img)) # 复用原始处理函数 process( input_docx=local_input, output_docx=output_docx_url, replacements=rep_pairs, image_replacements=img_pairs, color_keywords=color_keywords, ) abs_out = os.path.abspath(output_docx_url) return { "output_path": abs_out, "output_url": _build_output_url(abs_out), } finally: if tmp_input and os.path.exists(tmp_input): try: os.remove(tmp_input) except OSError: pass for p in tmp_images: if os.path.exists(p): try: os.remove(p) except OSError: pass if __name__ == "__main__": parser = argparse.ArgumentParser(description="DOCX MCP 服务器") parser.add_argument( "--transport", choices=["stdio", "http"], default="stdio", help="传输方式：stdio（本地）或 http（远程 HTTP /streamable-http）", ) parser.add_argument( "--host", default="0.0.0.0", help="HTTP 模式监听地址（默认 0.0.0.0）", ) parser.add_argument( "--port", type=int, default=8080, help="HTTP 模式监听端口（默认 8080）", ) args = parser.parse_args() if args.transport == "http": # HTTP 远程模式：通过 streamable-http 暴露 MCP 服务，端点 /mcp # 兼容当前 FastMCP 版本：通过 settings 配置 host/port，而不是传给 run() mcp.settings.host = args.host mcp.settings.port = args.port print(f"🚀 MCP HTTP 服务器启动中 → http://{args.host}:{args.port}/mcp") mcp.run(transport="streamable-http") else: # 本地 stdio 模式 print("🚀 MCP stdio 模式启动中（本地使用）") mcp.run(transport="stdio")