关于 LLM Gateway MCP Server
LLM Gateway MCP Server 的 README 与 Ultimate MCP Server 相同，描述了一个基于 MCP 的综合服务，向 AI Agent 提供多供应商 LLM 委派、浏览器自动化、文档处理、向量操作和认知记忆等能力。它适合构建多工具协作 Agent 的开发者，设计重点在于成本、性能与质量的平衡。
主要功能
- 01多供应商 LLM 委派与成本感知路由
- 02Playwright 浏览器自动化
- 03认知记忆系统（工作、情景、语义、程序记忆）
- 04文档分块、摘要与结构化数据提取
- 05带目录边界限制的文件系统操作
- 06Excel 表格自动化
适用场景
- 01将摘要或提取任务委派给成本较低的模型
- 02为 Agent 在多个步骤间保持状态与上下文
- 03自动化网页检索并汇总为报告
MCP 相关
该项目是一个 MCP 服务，将其工具和能力以标准 MCP 工具形式暴露给 Claude 等 MCP 客户端调用。
配置
本地服务的配置在各客户端中通用，复制后粘贴到客户端的 MCP 配置文件即可。
暂未收录安装方式
我们还没有获取到这个服务的安装包或远程地址，请按项目 README 中的说明安装。
工具
还没有获取到工具列表
本地服务需要在你的电脑上运行后才能读取工具，维护者暂未提供工具列表。
详细介绍
🧠 Ultimate MCP Server
一个功能全面的 Model Context Protocol（MCP）服务，为先进 AI Agent 提供数十种强大能力，用于认知增强、工具使用和智能编排
🤖 什么是 Ultimate MCP Server？
Ultimate MCP Server 是一个全面的 MCP 原生系统，可作为完整的 AI Agent 操作系统。它通过 Model Context Protocol 提供数十种强大能力，让先进 AI Agent 能够使用丰富的工具、认知系统和专业服务生态。
它不仅能够将复杂模型（例如 Claude 3.7 Sonnet）的任务智能委派给经济实惠的模型（例如 Gemini Flash 2.0 Lite），这只是其丰富功能的一部分。该服务提供对多个 LLM 提供商的统一访问，同时优化成本、性能和质量。
该系统集成了认知记忆系统、浏览器自动化、Excel 操作、数据库交互、文档处理、命令行实用工具、动态 API 集成、OCR 能力、向量操作、实体关系图、SQL 数据库交互、音频转录等功能。这些能力将 AI Agent 从对话界面转变为强大的自主系统，使其能够在数字环境中执行复杂的多步骤操作。
---## 🎯 愿景：完整的 AI Agent 操作系统
Ultimate MCP Server 的核心，是对 AI Agent 在数字环境中运行方式的一次根本性转变。它是一个面向 AI 的综合操作系统，提供：
- 🧠 统一的认知架构，支持持久记忆、推理和上下文感知
- ⚙️ 无缝访问数十种专业工具，涵盖网页浏览、文档处理、数据分析等领域
- 💻 直接提供系统级能力，支持文件系统操作、数据库交互和命令行实用工具
- 🔄 动态工作流能力，用于复杂的多步骤任务编排和执行
- 🌐 智能集成各种 LLM 提供商，并优化成本、质量和性能
- 🚀 提供高级向量操作、知识图谱和检索增强生成，增强 AI 能力
这种方式类似于复杂操作系统为应用程序提供硬件、服务和资源访问能力，但专为增强 AI Agent 而设计，使其获得超越自身原生能力的强大新功能。
🔌 MCP 原生架构
该服务完全基于 Model Context Protocol（MCP） 构建，专为与 Claude 等 AI Agent 协作而设计。所有功能都通过标准化的 MCP 工具提供，可由这些 Agent 直接调用，从而在 AI Agent 与丰富的能力、服务和外部系统生态之间构建无缝集成层。
🧬 核心用例：增强 AI Agent 与生态系统
Ultimate MCP Server 可将 Claude 3.7 Sonnet 等 AI Agent 转变为自主系统，使其能够在数字环境中执行复杂操作：
interacts with
┌─────────────┐ ────────────────────────► ┌───────────────────┐ ┌──────────────┐
│ Claude 3.7 │ │ Ultimate MCP │ ───────►│ LLM Providers│
│ (Agent) │ ◄──────────────────────── │ Server │ ◄───────│ External │
└─────────────┘ returns results └───────────────────┘ │ Systems │
│ └──────────────┘
▼
┌─────────────────────────────────────────────┐
│ Cognitive Memory Systems │
│ Web & Data: Browser, DB, RAG, Vector Search │
│ Documents: Excel, OCR, PDF, Filesystem │
│ Analysis: Entity Graphs, Classification │
│ Integration: APIs, CLI, Audio, Multimedia │
└─────────────────────────────────────────────┘
示例工作流：
- AI Agent 收到一项复杂任务，需要具备多种超出其原生能力的功能
- Agent 根据需要使用 Ultimate MCP Server 访问专用工具和服务
- Agent 可利用认知记忆系统，在各项操作中维持状态和上下文
- 研究、数据分析、文档创建和多媒体处理等复杂任务由此成为可能
- Agent 可编排多步骤工作流，以复杂的顺序组合使用各种工具
- 结果以标准 MCP 格式返回，使 Agent 能够理解并使用这些结果
- 一项重要优势是通过将合适的任务委派给效率更高的模型来优化成本
这种集成释放了变革性能力，使 AI Agent 能够自主完成复杂项目，同时智能地利用资源——包括在适当情况下使用专用工具和高性价比模型，可能节省 70-90% 的 API 成本。
💡 为什么使用 Ultimate MCP Server？
🧰 全面的 AI Agent 工具包
一个统一的枢纽，让高级 AI Agent 能够使用丰富的工具生态系统：
- 🌐 执行复杂的网页自动化任务（集成 Playwright）。
- 📊 深度集成地处理和分析 Excel 电子表格。
- 🧠 使用丰富的 认知记忆 系统，持久保存 Agent 状态。
- 💾 安全地访问 文件系统。
- 🗄️ 通过 SQL 操作访问 数据库。
- 🖼️ 使用 OCR 功能处理文档。
- 🔍 执行复杂的 向量搜索 和 RAG 操作。
- 🏷️ 使用专用的 文本处理 和 分类 功能。
- ⌨️ 使用 ripgrep、awk、sed、jq 等命令行工具。
- 🔌 动态集成外部 REST API。
- ✨ 使用 元工具 实现自我发现、优化和文档改进。
💵 成本优化
高级模型的 API 成本可能相当高。Ultimate MCP Server 可通过以下方式降低成本：
- 📉 将合适的任务路由给更便宜的模型（例如，每 1K 个 token 收费 $0.01，而非 $0.15）。
- ⚡ 实施 高级缓存（精确、语义、任务感知），避免重复调用 API。
- 💰 跟踪并 优化 各服务提供商的成本。
- 🧭 支持 成本感知型任务路由 决策。
- 🛠️ 使用专用的非大模型工具（文件系统、CLI 实用工具等）处理常规任务。
🌐 服务提供商抽象
通过统一接口避免受服务提供商锁定：
- 🔗 为 OpenAI、Anthropic（Claude）、Google（Gemini）、xAI（Grok）、DeepSeek、OpenRouter 和 本地 OpenAI 兼容服务（Ollama、llama.cpp、mistral.rs、vLLM、LM Studio）提供标准 API。
- 🏠 免费的本地推理：单个可配置的
local服务提供商可通过base_url连接任何 OpenAI 兼容的本地服务，并按 $0 成本计费，因此成本优化器会优先选择它来处理委派的工作。 - ⚙️ 统一处理参数和响应格式。
- 🔄 无需更改应用代码即可 切换服务提供商。
- 🛡️ 通过回退机制，避免服务提供商特定的故障和限制带来的影响。
📑 全面的文档和数据处理
高效处理文档和数据：
- ✂️ 将文档拆分为语义上有意义的 文本块。
- 🚀 跨多个模型 并行 处理文本块。
- 📊 从非结构化文本中提取 结构化数据（JSON、表格、键值对）。
- ✍️ 从长篇文本中生成 摘要 和见解。
- 🔁 转换格式（HTML 转 Markdown、文档转结构化数据）。
- 👁️ 对图像和 PDF 应用 OCR，并可选择使用大模型增强。
🚀 主要功能
🔌 MCP 协议集成
- 原生 MCP 服务：基于 Model Context Protocol 构建，可无缝集成 AI Agent。
- MCP 工具框架：所有功能均通过标准化的 MCP 工具提供，并配有清晰的架构定义。
- 工具组合：可通过依赖关系将工具组合到工作流中。
- 工具发现：支持动态列出工具并发现其能力，供 Agent 使用。
🤖 智能任务委派
- 任务路由：分析任务，并将其路由到合适的模型或专用工具。
- 提供商选择：根据任务要求、成本、质量或速度偏好选择提供商和模型。
- 成本与性能平衡：优化委派策略。
- 委派跟踪：监控委派模式、成本和结果（通过 Analytics）。
🌍 提供商集成
- 多提供商支持：通过单个可配置的
local提供商，全面支持 OpenAI、Anthropic、Google、DeepSeek、xAI（Grok）、OpenRouter，以及本地兼容 OpenAI 的服务（Ollama、llama.cpp、mistral.rs、vLLM、LM Studio）。架构可扩展。 - 免费本地推理：
local提供商的成本按 $0 计算，因此在配置了能力足够的本地模型后，智能委派／成本优化层会将对成本敏感的工作（摘要、提取、简单问答、格式整理）路由到你自己的硬件。 - 模型管理：处理不同模型的能力、上下文窗口和定价，并提供自动选择和回退机制。
💾 高级缓存
- 多级缓存：支持精确匹配、语义相似度和任务感知策略。
- 持久化缓存：基于磁盘的持久化（例如 DiskCache），并配有快速的内存访问层。
- 缓存分析：跟踪缓存命中率和预估成本节省。
📄 文档工具
- 智能分块：支持基于 token、语义边界检测和结构分析的方法。重叠长度可配置。
- 文档操作：摘要（段落、项目符号）、实体提取、问题生成和批处理。
📁 安全文件系统操作
- 路径管理：提供可靠的验证、规范化、符号链接安全检查和可配置的允许目录。
- 文件操作：支持读写和编码处理、智能文本编辑／替换以及元数据检索。
- 目录操作：支持创建、列出、树状视图以及安全移动／复制。
- 搜索功能：支持基于模式匹配和筛选的递归搜索。
- 安全重点：旨在防止目录遍历并强制执行边界限制。
✨ 自主工具文档优化器
- 自动改进：系统地分析、测试并优化 MCP 工具文档（文档字符串、架构定义、示例）。
- Agent 模拟：从大模型 Agent 的角度识别歧义。
- 自适应测试：生成并执行感知架构的测试用例。
- 失败分析：使用大模型集成来诊断文档缺陷。
- 迭代优化：持续提升文档质量。
- （更多详情请参阅专门章节）
🌐 使用 Playwright 进行浏览器自动化
- 全面控制：导航、点击、输入、抓取数据、截图、生成 PDF、文件上传／下载、执行 JS。
- 研究：自动跨搜索引擎执行搜索、提取结构化数据、监控网站。
- 综合整理：将多个网络来源的发现整合成报告。
🧠 认知与 Agent 记忆系统
- 记忆层级：工作记忆、情景记忆、语义记忆和程序记忆。
- 知识管理：存储／检索带有元数据、关系和重要性跟踪的记忆。
- 工作流跟踪：记录 Agent 操作、推理链、产物和依赖关系。
- 智能操作：记忆整合、反思生成、基于相关性的优化和衰减。
📊 Excel 电子表格自动化
- 直接操作：通过自然语言或结构化指令创建、修改和设置 Excel 文件格式。分析公式。
- 模板学习：从示例中学习、调整模板并应用格式模式。
- VBA 宏生成：根据指令生成 VBA 代码，实现复杂自动化。
🏗️ 结构化数据提取
- JSON 提取：提取结构化 JSON，并验证架构。
- 表格提取：以多种格式提取表格（JSON、CSV、Markdown）。
- 键值提取：提取简单的键值对。
- 语义架构推断：尝试从文本生成架构。
⚔️ 锦标赛模式
- 模型竞赛：针对代码或文本生成任务进行正面对比。
- 多模型评估：同时比较不同模型/提供商的输出。
- 性能指标：评估正确性、效率、风格等，并保存结果。
🗄️ SQL 数据库交互
- 查询执行：针对各种类型的数据库运行 SQL 查询（通过 SQLAlchemy 支持 SQLite、PostgreSQL 等）。
- 架构分析：分析架构并提出优化建议（使用 LLM）。
- 数据探索：浏览表格并查看其内容。
- 查询生成：根据自然语言描述生成 SQL。
🔗 实体关系图
- 实体提取：识别实体（人物、组织、地点等）。
- 关系映射：发现并映射实体之间的联系。
- 知识图谱构建：构建持久化图谱（例如，使用 NetworkX）。
- 图查询：通过图遍历或基于 LLM 的查询提取见解。
🔎 高级向量操作
- 语义搜索：使用向量嵌入查找相似内容。
- 向量存储集成：连接向量数据库或本地存储。
- 混合搜索：结合关键词搜索和语义搜索（例如，通过 Marqo 集成）。
- 批处理：为大型数据集高效生成嵌入并执行搜索。
📚 检索增强生成（RAG）
- 上下文生成：通过检索相关文档/文本块来增强提示词。
- 提升准确性：以提供的上下文为依据生成回答，减少幻觉。
- 工作流集成：无缝结合检索（向量/关键词搜索）与生成。策略可自定义。
🎙️ 音频转录
- 语音转文本：使用 Whisper 等模型将音频文件（例如，WAV、MP3）转换为文本。
- 说话人分离：识别不同的说话人（取决于模型/库是否支持）。
- 转录文本增强：使用 LLM 清理和整理转录文本。
- 多语言支持：根据所用转录模型支持多种语言。
🏷️ 文本分类
- 自定义分类器：应用文本分类模型（可能经过微调或使用零样本 LLM）。
- 多标签分类：分配多个类别。
- 置信度评分：提供分类概率。
- 批处理：高效分类大量文档。
👁️ OCR 工具
- PDF/图像提取：使用 Tesseract 或其他 OCR 引擎，并通过 LLM 进行校正/格式整理。
- 预处理：提供图像降噪、阈值处理和倾斜校正选项。
- 结构分析：提取 PDF 元数据和结构。
- 批处理：并发处理多个文件。
- （需要
ocr额外依赖：uv pip install -e ".[ocr]"）
📝 文本修订标记工具
- HTML 修订标记生成：显示文本/HTML 之间的差异（插入、删除、移动），并生成独立的 HTML 输出。
- 文档比较：比较各种格式，并通过直观的高亮显示差异。
🔄 HTML 转 Markdown
- 智能转换：检测内容类型，使用
readability-lxml、trafilatura、markdownify等库。 - 内容提取：过滤模板内容，保留结构（表格、链接）。
- Markdown 优化：清理并规范化输出。
📈 工作流优化工具
- 成本估算／比较：执行前估算成本，比较模型成本。
- 模型选择指南：根据任务、预算和性能需求推荐模型。
- 工作流执行引擎：运行包含依赖关系、并行执行和变量传递的多阶段流水线。
💻 本地文本处理工具（CLI 集成）
- 离线处理能力：安全地封装并公开
ripgrep（快速正则搜索）、awk（文本处理）、sed（流编辑器）、jq（JSON 处理）等命令行工具，作为 MCP 工具使用。无需 API 调用，即可在本地处理文本。
⏱️ 模型性能基准测试
- 实测指标：提供工具来测量不同提供方／模型的实际速度（tokens/sec）和延迟。
- 性能概况：根据真实场景下的性能生成对比报告。
- 数据驱动优化：利用基准测试数据指导路由决策。
📡 多种传输模式
- Streamable-HTTP（推荐）：现代 HTTP 传输，支持流式请求／响应正文，是基于 HTTP 的 MCP 客户端的最佳选择。
- Server-Sent Events（SSE）：采用服务器发送事件实现实时流式传输的旧版 HTTP 传输。
- 标准输入／输出（stdio）：用于嵌入式集成的直接进程通信。
- 实时流式传输：所有 HTTP 传输均支持 LLM 补全的逐 token 更新。
- 进度监控：跟踪长时间运行任务（分块、批处理）的进度。
- 事件驱动架构：订阅特定的服务事件。
✨ 多模型合成
- 对比分析：并排分析多个模型的输出。
- 响应合成：结合最佳内容，生成元响应，创建共识输出。
- 协作推理：实现由不同模型分别处理不同步骤的工作流。
🧩 扩展模型支持
- Grok 集成：原生支持 xAI 的 Grok。
- DeepSeek 支持：针对 DeepSeek 模型进行优化处理。
- OpenRouter 集成：使用 OpenRouter API 密钥即可访问丰富的模型。
- 本地／自托管集成：通过一个可配置的
local提供方，支持任何兼容 OpenAI 的本地服务端——Ollama、llama.cpp 的llama-server、mistral.rs、vLLM 和 LM Studio。将其指向base_url，即可在自己的硬件上运行免费（成本为 $0）的推理。 - Gemini 集成：全面支持 Google 的 Gemini 模型。
- Anthropic 集成：全面支持 Claude 模型，包括 Claude 3.5 Sonnet 和 Haiku。
- OpenAI 集成：全面支持 GPT-3.5、GPT-4.0 及更新的模型。
🔧 用于自我改进与动态集成的元工具
- 工具发现：Agent 可以查询可用工具、参数和说明（
list_tools）。 - 使用建议：获取 AI 驱动的工具选择／组合建议，以完成任务。
- 外部 API 集成：通过 OpenAPI 规范动态注册 REST API，使其端点可用作可调用的 MCP 工具（
register_api、call_dynamic_tool）。 - 文档生成：作为 Autonomous Refiner 功能的一部分。
📊 分析与报告
- 使用情况跟踪：按提供方／模型／工具监控 token 数量、成本、请求数、成功率／错误率。
- 实时监控：实时查看使用统计信息的仪表板或数据流。
- 详细报告：生成历史成本／使用情况报告、识别趋势并导出数据。
- 优化洞察：帮助识别高成本操作或低效模式。
📜 提示词模板与管理
- Jinja2 模板：使用变量、条件语句和包含项创建可复用的动态提示词。
- 提示词仓库：存储、检索、分类提示词，并对其进行版本控制。
- 元数据：为模板添加描述、作者信息和使用示例。
- 优化：测试并比较模板性能和令牌用量。
🛡️ 错误处理与韧性
- 智能重试：针对临时错误（速率限制、网络问题）自动进行指数退避重试。
- 回退机制：主提供商调用失败时，可配置回退提供商。
- 详细错误报告：捕获全面的错误上下文以便调试。
- 输入验证：针对常见问题（例如令牌限制、必需参数）进行预检。
⚙️ 系统功能
- 丰富的日志记录：通过
Rich输出多彩且信息丰富的控制台日志。 - 运行状况监控：提供
/healthz端点以进行就绪状态检查。 - 命令行界面：通过
umcpCLI 进行管理和交互。
📦 入门指南
🧪 安装
# Install uv (fast Python package manager) if you don't have it:
curl -LsSf https://astral.sh/uv/install.sh | sh
# Clone the repository
git clone https://github.com/Dicklesworthstone/ultimate_mcp_server.git
cd ultimate_mcp_server
# Create a virtual environment and install dependencies using uv:
uv venv --python 3.13
source .venv/bin/activate
uv lock --upgrade
uv sync --all-extras
注意：uv sync --all-extras 命令会安装项目中定义的所有可选扩展（例如 OCR、浏览器自动化、Excel）。如果只需要特定扩展，请调整项目依赖项并运行不带 --all-extras 的 uv sync。
⚙️ .env 配置
在克隆的仓库根目录中创建名为 .env 的文件。添加 API 密钥以及所需的配置覆盖项：
# --- API Keys (at least one provider required) ---
OPENAI_API_KEY=your_openai_sk-...
ANTHROPIC_API_KEY=your_anthropic_sk-...
GEMINI_API_KEY=your_google_ai_studio_key... # For Google AI Studio (Gemini API)
# Or use GOOGLE_APPLICATION_CREDENTIALS=/path/to/your/service-account-key.json for Vertex AI
DEEPSEEK_API_KEY=your_deepseek_key...
OPENROUTER_API_KEY=your_openrouter_key...
GROK_API_KEY=your_grok_key... # For Grok via xAI API
# --- Local / Self-Hosted Providers (OpenAI-compatible, FREE inference) ---
# One generic provider covers Ollama, llama.cpp (llama-server), mistral.rs, vLLM, and LM Studio.
# No API key is required by most local servers; LOCAL_LLM_API_KEY is optional.
# LOCAL_LLM_BASE_URL=http://localhost:11434/v1 # Default (Ollama). Examples:
# llama.cpp / mistral.rs / vLLM : http://localhost:8000/v1
# LM Studio : http://localhost:1234/v1
# LOCAL_LLM_DEFAULT_MODEL=llama3.1:8b # Model name as served by your local backend
# LOCAL_LLM_API_KEY= # Optional; most local servers ignore it
# LOCAL_LLM_REQUEST_TIMEOUT=30 # Optional request timeout in seconds
# LOCAL_LLM_ENABLED=true # Optional; set false to disable the local provider
# --- Server Configuration (Defaults shown) ---
GATEWAY_SERVER_PORT=8013
GATEWAY_SERVER_HOST=127.0.0.1 # Change to 0.0.0.0 to listen on all interfaces (needed for Docker/external access)
# GATEWAY_API_PREFIX=/
# --- Logging Configuration (Defaults shown) ---
LOG_LEVEL=INFO # DEBUG, INFO, WARNING, ERROR, CRITICAL
USE_RICH_LOGGING=true # Set to false for plain text logs
# --- Cache Configuration (Defaults shown) ---
GATEWAY_CACHE_ENABLED=true
GATEWAY_CACHE_TTL=86400 # Default Time-To-Live in seconds (24 hours)
# GATEWAY_CACHE_TYPE=memory # Options might include 'memory', 'redis', 'diskcache' (check implementation)
# GATEWAY_CACHE_MAX_SIZE=1000 # Example: Max number of items for memory cache
# GATEWAY_CACHE_DIR=./.cache # Directory for disk cache storage
# --- Provider Timeouts & Retries (Defaults shown) ---
# GATEWAY_PROVIDER_TIMEOUT=120 # Default timeout in seconds for API calls
# GATEWAY_PROVIDER_MAX_RETRIES=3 # Default max retries on failure
# --- Provider-Specific Configuration ---
# GATEWAY_OPENAI_DEFAULT_MODEL=gpt-4.1-mini # Customize default model
# GATEWAY_ANTHROPIC_DEFAULT_MODEL=claude-3-5-sonnet-20241022 # Customize default model
# GATEWAY_GEMINI_DEFAULT_MODEL=gemini-2.0-pro # Customize default model
# --- Tool Specific Config (Examples) ---
# FILESYSTEM__ALLOWED_DIRECTORIES=["/path/to/safe/dir1","/path/to/safe/dir2"] # For Filesystem tools (JSON array)
# GATEWAY_AGENT_MEMORY_DB_PATH=unified_agent_memory.db # Path for agent memory database
# GATEWAY_PROMPT_TEMPLATES_DIR=./prompt_templates # Directory for prompt templates
▶️ 运行
确保虚拟环境已激活（source .venv/bin/activate）。
# Start the MCP server with all registered tools found
umcp run
# Start the server including only specific tools
umcp run --include-tools completion chunk_document read_file write_file
# Start the server excluding specific tools
umcp run --exclude-tools browser_init browser_navigate research_and_synthesize_report
# Start with Docker (ensure .env file exists in the project root or pass environment variables)
docker compose up --build # Add --build the first time or after changes
启动后，服务通常可通过 http://localhost:8013 访问（也可以使用 .env 或命令行中配置的主机和端口）。你应该能看到日志输出，显示服务已启动以及已注册的工具。
💻 命令行界面（CLI）
Ultimate MCP Server 通过 umcp 命令提供功能强大的命令行界面（CLI），可用于管理服务、与大模型提供商交互、测试功能以及查看示例。本节详细介绍所有可用命令及其选项。
🌟 全局选项
umcp 命令支持以下全局选项：
umcp --version # Display version information
🚀 服务管理
启动服务
run 命令使用指定选项启动 Ultimate MCP Server：
# Basic server start with default settings from .env
umcp run
# Run on a specific host (-h) and port (-p)
umcp run -h 0.0.0.0 -p 9000
# Run with multiple worker processes (-w)
umcp run -w 4
# Enable debug logging (-d)
umcp run -d
# Use stdio transport (-t)
umcp run -t stdio
# Use streamable-http transport (recommended for HTTP clients)
umcp run -t shttp
# Run only with specific tools (no shortcut for --include-tools)
umcp run --include-tools completion chunk_document read_file write_file
# Run with all tools except certain ones (no shortcut for --exclude-tools)
umcp run --exclude-tools browser_init browser_navigate
输出示例：
┌─ Starting Ultimate MCP Server ───────────────────┐
│ Host: 0.0.0.0 │
│ Port: 9000 │
│ Workers: 4 │
│ Transport mode: streamable-http │
└────────────────────────────────────────────────┘
INFO: Started server process [12345]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:9000 (Press CTRL+C to quit)
可用选项：
-h, --host：要绑定服务的主机或 IP 地址（默认值：来自 .env）-p, --port：要监听的端口（默认值：来自 .env）-w, --workers：要启动的工作进程数（默认值：来自 .env）-t, --transport-mode：服务通信的传输模式（'shttp'表示 streamable-http，也可以是'sse'或'stdio'，默认值：shttp）-d, --debug：启用调试日志--include-tools：要包含的工具名称列表（逗号分隔）--exclude-tools：要排除的工具名称列表（逗号分隔）
🔌 提供商管理
列出提供商
providers 命令会显示已配置大模型提供商的信息：
# List all configured providers
umcp providers
# Check API keys (-c) for all configured providers
umcp providers -c
# List available models (no shortcut for --models)
umcp providers --models
# Check keys and list models
umcp providers -c --models
输出示例：
┌─ LLM Providers ──────────────────────────────────────────────────┐
│ Provider Status Default Model API Key │
├───────────────────────────────────────────────────────────────────┤
│ openai ✓ gpt-4.1-mini sk-...5vX [VALID] │
│ anthropic ✓ claude-3-5-sonnet-20241022 sk-...Hr [VALID] │
│ gemini ✓ gemini-2.0-pro [VALID] │
│ deepseek ✗ deepseek-chat [NOT CONFIGURED] │
│ openrouter ✓ -- [VALID] │
│ grok ✓ grok-1 [VALID] │
└───────────────────────────────────────────────────────────────────┘
使用 --models：
OPENAI MODELS:
- gpt-4.1-mini
- gpt-4o
- gpt-4-0125-preview
- gpt-3.5-turbo
ANTHROPIC MODELS:
- claude-3-5-sonnet-20241022
- claude-3-5-haiku-20241022
- claude-3-opus-20240229
...
可用选项：
-c, --check：检查所有已配置提供商的 API 密钥--models：列出每个提供商可用的模型
测试提供商
test 命令可用于测试特定提供商：
# Test the default OpenAI model with a simple prompt
umcp test openai
# Test a specific model (--model) with a custom prompt (--prompt)
umcp test anthropic --model claude-3-5-haiku-20241022 --prompt "Write a short poem about coding."
# Test Gemini with a different prompt
umcp test gemini --prompt "What are three interesting AI research papers from 2024?"
输出示例：
Testing provider 'anthropic'...
Provider: anthropic
Model: claude-3-5-haiku-20241022
Prompt: Write a short poem about coding.
❯ Response:
Code flows like water,
Logic cascades through the mind—
Bugs bloom like flowers.
Tokens: 13 input, 19 output
Cost: $0.00006
Response time: 0.82s
可用选项：
--model：要测试的模型 ID（默认为该提供商的默认模型）--prompt：要发送的提示词文本（默认值："Hello, world!"）
⚡ 直接生成文本
complete 命令可让你直接通过 CLI 生成文本：
# Generate text with default provider (OpenAI) using a prompt (--prompt)
umcp complete --prompt "Write a concise explanation of quantum computing."
# Specify a provider (--provider) and model (--model)
umcp complete --provider anthropic --model claude-3-5-sonnet-20241022 --prompt "What are the key differences between Rust and Go?"
# Use a system prompt (--system)
umcp complete --provider openai --model gpt-4o --system "You are an expert programmer..." --prompt "Explain dependency injection."
# Stream the response token by token (-s)
umcp complete --provider openai --prompt "Count from 1 to 10." -s
# Adjust temperature (--temperature) and token limit (--max-tokens)
umcp complete --provider gemini --temperature 1.2 --max-tokens 250 --prompt "Generate a creative sci-fi story opening."
# Read prompt from stdin (no --prompt needed)
echo "Tell me about space exploration." | umcp complete
输出示例：
Quantum computing uses quantum bits (qubits) that can exist in multiple states simultaneously, unlike classical bits (0 or 1). This quantum superposition, along with entanglement, allows quantum computers to process vast amounts of information in parallel, potentially solving certain complex problems exponentially faster than classical computers. Applications include cryptography, materials science, and optimization problems.
Tokens: 13 input, 72 output
Cost: $0.00006
Response time: 0.37s
可用选项：
--provider：要使用的提供商（默认：openai）--model：模型 ID（默认为提供商的默认模型）--prompt：提示词文本（如果未提供，则从标准输入读取）--temperature：采样温度（0.0-2.0，默认：0.7）--max-tokens：要生成的最大 token 数--system：支持此功能的提供商所用的系统提示词-s, --stream：逐个 token 流式传输响应
💾 缓存管理
cache 命令可让你查看或清除请求缓存：
# Show cache status (default action)
umcp cache
# Explicitly show status (no shortcut for --status)
umcp cache --status
# Clear the cache (no shortcut for --clear, with confirmation prompt)
umcp cache --clear
# Show stats and clear the cache in one command
umcp cache --status --clear
输出示例：
Cache Status:
Backend: memory
Enabled: True
Items: 127
Hit rate: 73.2%
Estimated savings: $1.47
可用选项：
--status：显示缓存状态（未设置其他标志时默认启用）--clear：清除缓存（会提示确认）
📊 基准测试
benchmark 命令可让你比较各提供商的性能和成本：
# Run default benchmark (3 runs per provider)
umcp benchmark
# Benchmark only specific providers
umcp benchmark --providers openai,anthropic
# Benchmark with specific models
umcp benchmark --providers openai,anthropic --models gpt-4o,claude-3.5-sonnet
# Use a custom prompt and more runs (-r)
umcp benchmark --prompt "Explain the process of photosynthesis in detail." -r 5
输出示例：
┌─ Benchmark Results ───────────────────────────────────────────────────────┐
│ Provider Model Avg Time Tokens Cost Tokens/sec │
├──────────────────────────────────────────────────────────────────────────┤
│ openai gpt-4.1-mini 0.47s 76 / 213 $0.00023 454 │
│ anthropic claude-3-5-haiku 0.52s 76 / 186 $0.00012 358 │
│ gemini gemini-2.0-pro 0.64s 76 / 201 $0.00010 314 │
│ deepseek deepseek-chat 0.71s 76 / 195 $0.00006 275 │
└──────────────────────────────────────────────────────────────────────────┘
可用选项：
--providers：要进行基准测试的提供商列表（默认：所有已配置的提供商）--models：要进行基准测试的模型 ID（默认为各提供商的默认模型）--prompt：要使用的提示词文本（默认：内置基准测试提示词）-r, --runs：每个提供商/模型的运行次数（默认：3）
🧰 工具管理
tools 命令会列出可用工具，也可按类别筛选：
# List all tools
umcp tools
# List tools in a specific category
umcp tools --category document
# Show related example scripts
umcp tools --examples
输出示例：
┌─ Ultimate MCP Server Tools ─────────────────────────────────────────┐
│ Category Tool Example Script │
├──────────────────────────────────────────────────────────────────────┤
│ completion generate_completion simple_completion_demo.py │
│ completion stream_completion simple_completion_demo.py │
│ completion chat_completion claude_integration_demo.py│
│ document summarize_document document_processing.py │
│ document chunk_document document_processing.py │
│ extraction extract_json advanced_extraction_demo.py│
│ filesystem read_file filesystem_operations_demo.py│
└──────────────────────────────────────────────────────────────────────┘
Tip: Run examples using the command:
umcp examples <example_name>
可用选项：
--category：按类别筛选工具--examples：在工具旁显示示例脚本
📚 示例管理
examples 命令可让你列出并运行示例脚本：
# List all example scripts (default action)
umcp examples
# Explicitly list example scripts (-l)
umcp examples -l
# Run a specific example
umcp examples rag_example.py
# Can also run by just the name without extension
umcp examples rag_example
列出示例时的输出：
┌─ Ultimate MCP Server Example Scripts ─────────────────────────────────┐
│ Category Example Script │
├────────────────────────────────────────────────────────────────────────┤
│ text-generation simple_completion_demo.py │
│ text-generation claude_integration_demo.py │
│ document-processing document_processing.py │
│ search-and-retrieval rag_example.py │
│ browser-automation browser_automation_demo.py │
└────────────────────────────────────────────────────────────────────────┘
Run an example:
umcp examples <example_name>
运行示例时：
Running example: rag_example.py
Creating vector knowledge base 'demo_kb'...
Adding sample documents...
Retrieving context for query: "What are the benefits of clean energy?"
Generated response:
Based on the retrieved context, clean energy offers several benefits:
...
可用选项：
-l, --list：仅列出示例脚本--category：按类别筛选示例
🔎 获取帮助
每个命令都有详细的帮助信息：
# General help
umcp --help
# Help for a specific command
umcp run --help
umcp providers --help
umcp complete --help
输出示例：
Usage: umcp [OPTIONS] COMMAND [ARGS]...
Ultimate MCP Server: Multi-provider LLM management server
Unified CLI to run your server, manage providers, and more.
Options:
--version, -v Show the application version and exit.
--help Show this message and exit.
Commands:
run Run the Ultimate MCP Server
providers List Available Providers
test Test a Specific Provider
complete Generate Text Completion
cache Cache Management
benchmark Benchmark Providers
tools List Available Tools
examples Run or List Example Scripts
特定命令的帮助：
Usage: umcp run [OPTIONS]
Run the Ultimate MCP Server
Start the server with optional overrides.
Examples:
umcp run -h 0.0.0.0 -p 8000 -w 4 -t sse
umcp run -d
Options:
-h, --host TEXT Host or IP address to bind the server to.
Defaults from config.
-p, --port INTEGER Port to listen on. Defaults from config.
-w, --workers INTEGER Number of worker processes to spawn.
Defaults from config.
-t, --transport-mode [shttp|sse|stdio]
Transport mode for server communication (-t
shortcut). Options: 'shttp' (streamable-http,
recommended), 'sse', or 'stdio'.
-d, --debug Enable debug logging for detailed output (-d
shortcut).
--include-tools TEXT List of tool names to include when running
the server.
--exclude-tools TEXT List of tool names to exclude when running
the server.
--help Show this message and exit.
🧪 使用示例
本节提供 Python 示例，演示 MCP 客户端（例如使用 mcp-client 的应用）或 Claude 等 Agent 如何与正在运行的 Ultimate MCP Server 实例提供的工具交互。
注意：这些示例假设你已安装 mcp-client（pip install mcp-client），并且 Ultimate MCP Server 正在 http://localhost:8013 运行。
（为确保内容完整，以下保留了原始输入中的详细代码块。）
基本补全
import asyncio
from mcp.client import Client
async def basic_completion_example():
client = Client("http://localhost:8013")
response = await client.tools.completion(
prompt="Write a short poem about a robot learning to dream.",
provider="openai",
model="gpt-4.1-mini",
max_tokens=100,
temperature=0.7
)
if response["success"]:
print(f"Completion: {response['completion']}")
print(f"Cost: ${response['cost']:.6f}")
else:
print(f"Error: {response['error']}")
await client.close()
# if __name__ == "__main__": asyncio.run(basic_completion_example())
Claude 使用 Ultimate MCP Server 分析文档（委派）
import asyncio
from mcp.client import Client
async def document_analysis_example():
# Assume Claude identifies a large document needing processing
client = Client("http://localhost:8013")
document = "... large document content ..." * 100 # Placeholder for large content
print("Delegating document chunking...")
# Step 1: Claude delegates document chunking (often a local, non-LLM task on server)
chunks_response = await client.tools.chunk_document(
document=document,
chunk_size=1000, # Target tokens per chunk
overlap=100, # Token overlap
method="semantic" # Use semantic chunking if available
)
if not chunks_response["success"]:
print(f"Chunking failed: {chunks_response['error']}")
await client.close()
return
print(f"Document divided into {chunks_response['chunk_count']} chunks.")
# Step 2: Claude delegates summarization of each chunk to a cheaper model
summaries = []
total_cost = 0.0
print("Delegating chunk summarization to gemini-2.0-flash-lite...")
for i, chunk in enumerate(chunks_response["chunks"]):
# Use Gemini Flash (much cheaper than Claude or GPT-4o) via the server
summary_response = await client.tools.summarize_document(
document=chunk,
provider="gemini", # Explicitly delegate to Gemini via server
model="gemini-2.0-flash-lite",
format="paragraph",
max_length=150 # Request a concise summary
)
if summary_response["success"]:
summaries.append(summary_response["summary"])
cost = summary_response.get("cost", 0.0)
total_cost += cost
print(f" Processed chunk {i+1}/{chunks_response['chunk_count']} summary. Cost: ${cost:.6f}")
else:
print(f" Chunk {i+1} summarization failed: {summary_response['error']}")
print("\nDelegating entity extraction to gpt-4.1-mini...")
# Step 3: Claude delegates entity extraction for the whole document to another cheap model
entities_response = await client.tools.extract_entities(
document=document, # Process the original document
entity_types=["person", "organization", "location", "date", "product"],
provider="openai", # Delegate to OpenAI's cheaper model
model="gpt-4.1-mini"
)
if entities_response["success"]:
cost = entities_response.get("cost", 0.0)
total_cost += cost
print(f"Extracted entities. Cost: ${cost:.6f}")
extracted_entities = entities_response['entities']
# Claude would now process these summaries and entities using its advanced capabilities
print(f"\nClaude can now use {len(summaries)} summaries and {len(extracted_entities)} entity groups.")
else:
print(f"Entity extraction failed: {entities_response['error']}")
print(f"\nTotal estimated delegation cost for sub-tasks: ${total_cost:.6f}")
# Claude might perform final synthesis using the collected results
final_synthesis_prompt = f"""
Synthesize the key information from the following summaries and entities extracted from a large document.
Focus on the main topics, key people involved, and significant events mentioned.
Summaries:
{' '.join(summaries)}
Entities:
{extracted_entities}
Provide a concise final report.
"""
# This final step would likely use Claude itself (not shown here)
await client.close()
# if __name__ == "__main__": asyncio.run(document_analysis_example())
用于研究的浏览器自动化
import asyncio
from mcp.client import Client
async def browser_research_example():
client = Client("http://localhost:8013")
print("Starting browser-based research task...")
# This tool likely orchestrates multiple browser actions (search, navigate, scrape)
# and uses an LLM (specified or default) for synthesis.
result = await client.tools.research_and_synthesize_report(
topic="Latest advances in AI-powered drug discovery using graph neural networks",
instructions={
"search_query": "graph neural networks drug discovery 2024 research",
"search_engines": ["google", "duckduckgo"], # Use multiple search engines
"urls_to_include": ["nature.com", "sciencemag.org", "arxiv.org", "pubmed.ncbi.nlm.nih.gov"], # Prioritize these domains
"max_urls_to_process": 7, # Limit the number of pages to visit/scrape
"min_content_length": 500, # Ignore pages with very little content
"focus_areas": ["novel molecular structures", "binding affinity prediction", "clinical trial results"], # Guide the synthesis
"report_format": "markdown", # Desired output format
"report_length": "detailed", # comprehensive, detailed, summary
"llm_model": "anthropic/claude-3-5-sonnet-20241022" # Specify LLM for synthesis
}
)
if result["success"]:
print("\nResearch report generated successfully!")
print(f"Processed {len(result.get('extracted_data', []))} sources.")
print(f"Total processing time: {result.get('processing_time', 'N/A'):.2f}s")
print(f"Estimated cost: ${result.get('total_cost', 0.0):.6f}") # Includes LLM synthesis cost
print("\n--- Research Report ---")
print(result['report'])
print("-----------------------")
else:
print(f"\nBrowser research failed: {result.get('error', 'Unknown error')}")
if 'details' in result: print(f"Details: {result['details']}")
await client.close()
# if __name__ == "__main__": asyncio.run(browser_research_example())
认知记忆系统的使用
import asyncio
from mcp.client import Client
import uuid
async def cognitive_memory_example():
client = Client("http://localhost:8013")
# Generate a unique ID for this session/workflow if not provided
workflow_id = str(uuid.uuid4())
print(f"Using Workflow ID: {workflow_id}")
print("\nCreating a workflow context...")
# Create a workflow context to group related memories and actions
workflow_response = await client.tools.create_workflow(
workflow_id=workflow_id,
title="Quantum Computing Investment Analysis",
description="Analyzing the impact of quantum computing on financial markets.",
goal="Identify potential investment opportunities or risks."
)
if not workflow_response["success"]: print(f"Error creating workflow: {workflow_response['error']}")
print("\nRecording an agent action...")
# Record the start of a research action
action_response = await client.tools.record_action_start(
workflow_id=workflow_id,
action_type="research",
title="Initial literature review on quantum algorithms in finance",
reasoning="Need to understand the current state-of-the-art before assessing impact."
)
action_id = action_response.get("action_id") if action_response["success"] else None
if not action_id: print(f"Error starting action: {action_response['error']}")
print("\nStoring facts in semantic memory...")
# Store some key facts discovered during research
memory1 = await client.tools.store_memory(
workflow_id=workflow_id,
content="Shor's algorithm can break RSA encryption, posing a threat to current financial security.",
memory_type="fact", memory_level="semantic", importance=9.0,
tags=["quantum_algorithm", "cryptography", "risk", "shor"]
)
memory2 = await client.tools.store_memory(
workflow_id=workflow_id,
content="Quantum annealing (e.g., D-Wave) shows promise for portfolio optimization problems.",
memory_type="fact", memory_level="semantic", importance=7.5,
tags=["quantum_computing", "finance", "optimization", "annealing"]
)
if memory1["success"]: print(f"Stored memory ID: {memory1['memory_id']}")
if memory2["success"]: print(f"Stored memory ID: {memory2['memory_id']}")
print("\nStoring an observation (episodic memory)...")
# Store an observation from a specific event/document
obs_memory = await client.tools.store_memory(
workflow_id=workflow_id,
content="Read Nature article (doi:...) suggesting experimental quantum advantage in a specific financial modeling task.",
memory_type="observation", memory_level="episodic", importance=8.0,
source="Nature Article XYZ", timestamp="2024-07-20T10:00:00Z", # Example timestamp
tags=["research_finding", "publication", "finance_modeling"]
)
if obs_memory["success"]: print(f"Stored episodic memory ID: {obs_memory['memory_id']}")
print("\nSearching for relevant memories...")
# Search for memories related to financial risks
search_results = await client.tools.hybrid_search_memories(
workflow_id=workflow_id,
query="What are the financial risks associated with quantum computing?",
top_k=5, memory_type="fact", # Search for facts first
semantic_weight=0.7, keyword_weight=0.3 # Example weighting for hybrid search
)
if search_results["success"]:
print(f"Found {len(search_results['results'])} relevant memories:")
for res in search_results["results"]:
print(f" - Score: {res['score']:.4f}, ID: {res['memory_id']}, Content: {res['content'][:80]}...")
else:
print(f"Memory search failed: {search_results['error']}")
print("\nGenerating a reflection based on stored memories...")
# Generate insights or reflections based on the accumulated knowledge in the workflow
reflection_response = await client.tools.generate_reflection(
workflow_id=workflow_id,
reflection_type="summary_and_next_steps", # e.g., insights, risks, opportunities
context_query="Summarize the key findings about quantum finance impact and suggest next research actions."
)
if reflection_response["success"]:
print("Generated Reflection:")
print(reflection_response["reflection"])
else:
print(f"Reflection generation failed: {reflection_response['error']}")
# Mark the action as completed (assuming research phase is done)
if action_id:
print("\nCompleting the research action...")
await client.tools.record_action_end(
workflow_id=workflow_id, action_id=action_id, status="completed",
outcome="Gathered initial understanding of quantum algorithms in finance and associated risks."
)
await client.close()
# if __name__ == "__main__": asyncio.run(cognitive_memory_example())
Excel 电子表格自动化
import asyncio
from mcp.client import Client
import os
async def excel_automation_example():
client = Client("http://localhost:8013")
output_dir = "excel_outputs"
os.makedirs(output_dir, exist_ok=True)
output_path = os.path.join(output_dir, "financial_model.xlsx")
print(f"Requesting creation of Excel financial model at {output_path}...")
# Example: Create a financial model using natural language instructions
create_result = await client.tools.excel_execute(
instruction="Create a simple 3-year financial projection.\n"
"Sheet name: 'Projections'.\n"
"Columns: Year 1, Year 2, Year 3.\n"
"Rows: Revenue, COGS, Gross Profit, Operating Expenses, Net Income.\n"
"Data: Start Revenue at $100,000, grows 20% annually.\n"
"COGS is 40% of Revenue.\n"
"Operating Expenses start at $30,000, grow 10% annually.\n"
"Calculate Gross Profit (Revenue - COGS) and Net Income (Gross Profit - OpEx).\n"
"Format currency as $#,##0. Apply bold headers and add a light blue fill to the header row.",
file_path=output_path, # Server needs write access to this path/directory if relative
operation_type="create", # create, modify, analyze, format
# sheet_name="Projections", # Can specify sheet if modifying
# cell_range="A1:D6", # Can specify range
show_excel=False # Run Excel in the background (if applicable on the server)
)
if create_result["success"]:
print(f"Excel creation successful: {create_result['message']}")
print(f"File saved at: {create_result.get('output_file_path', output_path)}") # Confirm output path
# Example: Modify the created file - add a chart
print("\nRequesting modification: Add a Revenue chart...")
modify_result = await client.tools.excel_execute(
instruction="Add a column chart showing Revenue for Year 1, Year 2, Year 3. "
"Place it below the table. Title the chart 'Revenue Projection'.",
file_path=output_path, # Use the previously created file
operation_type="modify",
sheet_name="Projections" # Specify the sheet to modify
)
if modify_result["success"]:
print(f"Excel modification successful: {modify_result['message']}")
print(f"File updated at: {modify_result.get('output_file_path', output_path)}")
else:
print(f"Excel modification failed: {modify_result['error']}")
else:
print(f"Excel creation failed: {create_result['error']}")
if 'details' in create_result: print(f"Details: {create_result['details']}")
# Example: Analyze formulas (if the tool supports it)
# analysis_result = await client.tools.excel_analyze_formulas(...)
await client.close()
# if __name__ == "__main__": asyncio.run(excel_automation_example())
多提供商比较
import asyncio
from mcp.client import Client
async def multi_provider_completion_example():
client = Client("http://localhost:8013")
prompt = "Explain the concept of 'Chain of Thought' prompting for Large Language Models."
print(f"Requesting completions for prompt: '{prompt}' from multiple providers...")
# Request the same prompt from different models/providers
multi_response = await client.tools.multi_completion(
prompt=prompt,
providers=[
{"provider": "openai", "model": "gpt-4.1-mini", "temperature": 0.5},
{"provider": "anthropic", "model": "claude-3-5-sonnet-20241022", "temperature": 0.5},
{"provider": "gemini", "model": "gemini-2.0-pro", "temperature": 0.5},
# {"provider": "deepseek", "model": "deepseek-chat", "temperature": 0.5}, # Add others if configured
],
# Common parameters applied to all if not specified per provider
max_tokens=300
)
if multi_response["success"]:
print("\n--- Multi-completion Results ---")
total_cost = multi_response.get("total_cost", 0.0)
print(f"Total Estimated Cost: ${total_cost:.6f}\n")
for provider_key, result in multi_response["results"].items():
print(f"--- Provider: {provider_key} ---")
if result["success"]:
print(f" Model: {result.get('model', 'N/A')}")
print(f" Cost: ${result.get('cost', 0.0):.6f}")
print(f" Tokens: Input={result.get('input_tokens', 'N/A')}, Output={result.get('output_tokens', 'N/A')}")
print(f" Completion:\n{result['completion']}\n")
else:
print(f" Error: {result['error']}\n")
print("------------------------------")
# An agent could now analyze these responses for consistency, detail, accuracy etc.
else:
print(f"\nMulti-completion request failed: {multi_response['error']}")
await client.close()
# if __name__ == "__main__": asyncio.run(multi_provider_completion_example())
成本优化工作流执行
import asyncio
from mcp.client import Client
async def optimized_workflow_example():
client = Client("http://localhost:8013")
# Example document to process through the workflow
document_content = """
Project Alpha Report - Q3 2024
Lead: Dr. Evelyn Reed ([email protected])
Status: On Track
Budget: $50,000 remaining. Spent $25,000 this quarter.
Key Findings: Successful prototype development (v0.8). User testing feedback positive.
Next Steps: Finalize documentation, prepare for Q4 deployment. Target date: 2024-11-15.
Risks: Potential delay due to supplier issues for component X. Mitigation plan in place.
"""
print("Defining a multi-stage workflow...")
# Define a workflow with stages, dependencies, and provider preferences
# Use ${stage_id.output_key} to pass outputs between stages
workflow_definition = [
{
"stage_id": "summarize_report",
"tool_name": "summarize_document",
"params": {
"document": document_content,
"format": "bullet_points",
"max_length": 100,
# Let the server choose a cost-effective model for summarization
"provider_preference": "cost", # 'cost', 'quality', 'speed', or specific like 'openai/gpt-4.1-mini'
}
# No 'depends_on', runs first
# Default output key is 'summary' for this tool, access via ${summarize_report.summary}
},
{
"stage_id": "extract_key_info",
"tool_name": "extract_json", # Use JSON extraction for structured data
"params": {
"document": document_content,
"json_schema": {
"type": "object",
"properties": {
"project_lead": {"type": "string"},
"lead_email": {"type": "string", "format": "email"},
"status": {"type": "string"},
"budget_remaining": {"type": "string"},
"next_milestone_date": {"type": "string", "format": "date"}
},
"required": ["project_lead", "status", "next_milestone_date"]
},
# Prefer a model known for good structured data extraction, balancing cost
"provider_preference": "quality", # Prioritize quality for extraction
"preferred_models": ["openai/gpt-4o", "anthropic/claude-3-5-sonnet-20241022"] # Suggest specific models
}
},
{
"stage_id": "generate_follow_up_questions",
"tool_name": "generate_qa", # Assuming a tool that generates questions
"depends_on": ["summarize_report"], # Needs the summary first
"params": {
# Use the summary from the first stage as input
"document": "${summarize_report.summary}",
"num_questions": 3,
"provider_preference": "speed" # Use a fast model for question generation
}
# Default output key 'qa_pairs', access via ${generate_follow_up_questions.qa_pairs}
}
]
print("Executing the optimized workflow...")
# Execute the workflow - the server handles dependencies and model selection
results = await client.tools.execute_optimized_workflow(
workflow=workflow_definition
# Can also pass initial documents if workflow steps reference 'original_document'
# documents = {"report.txt": document_content}
)
if results["success"]:
print("\nWorkflow executed successfully!")
print(f" Total processing time: {results.get('processing_time', 'N/A'):.2f}s")
print(f" Total estimated cost: ${results.get('total_cost', 0.0):.6f}\n")
print("--- Stage Outputs ---")
for stage_id, output in results.get("stage_outputs", {}).items():
print(f"Stage: {stage_id}")
if output["success"]:
print(f" Provider/Model Used: {output.get('provider', 'N/A')}/{output.get('model', 'N/A')}")
print(f" Cost: ${output.get('cost', 0.0):.6f}")
print(f" Output: {output.get('result', 'N/A')}") # Access the primary result
# You might access specific keys like output.get('result', {}).get('summary') etc.
else:
print(f" Error: {output.get('error', 'Unknown error')}")
print("-" * 20)
else:
print(f"\nWorkflow execution failed: {results.get('error', 'Unknown error')}")
if 'details' in results: print(f"Details: {results['details']}")
await client.close()
# if __name__ == "__main__": asyncio.run(optimized_workflow_example())
实体关系图示例
import asyncio
from mcp.client import Client
# import networkx as nx # To process the graph data if needed
# import matplotlib.pyplot as plt # To visualize
async def entity_graph_example():
client = Client("http://localhost:8013")
document_text = """
Meta Platforms, Inc., led by CEO Mark Zuckerberg, announced a partnership with IBM
on developing new AI hardware accelerators. The collaboration aims to challenge Nvidia's dominance.
IBM, headquartered in Armonk, New York, brings its deep expertise in semiconductor design.
The project, codenamed 'Synergy', is expected to yield results by late 2025.
"""
print("Extracting entity relationships from text...")
# Request extraction of entities and their relationships
entity_graph_response = await client.tools.extract_entity_relations(
document=document_text,
entity_types=["organization", "person", "location", "date", "project"], # Specify desired entity types
relationship_types=["led_by", "partnership_with", "aims_to_challenge", "headquartered_in", "expected_by"], # Specify relationship types
# Optional parameters:
# provider_preference="quality", # Choose model strategy
# llm_model="anthropic/claude-3-5-sonnet-20241022", # Suggest a specific model
include_visualization=False # Set True to request image data if tool supports it
)
if entity_graph_response["success"]:
print("Entity relationship extraction successful.")
print(f"Estimated Cost: ${entity_graph_response.get('cost', 0.0):.6f}")
# The graph data might be in various formats (e.g., node-link list, adjacency list)
graph_data = entity_graph_response.get("graph_data")
print("\n--- Graph Data (Nodes & Edges) ---")
print(graph_data)
print("------------------------------------")
# Example: Query the extracted graph using another tool or LLM call
# (Assuming a separate query tool or using a general completion tool)
print("\nQuerying the extracted graph (example)...")
query_prompt = f"""
Based on the following graph data representing relationships extracted from a text:
{graph_data}
Answer the question: Who is the CEO of Meta Platforms, Inc.?
"""
query_response = await client.tools.completion(
prompt=query_prompt, provider="openai", model="gpt-4.1-mini", max_tokens=50
)
if query_response["success"]:
print(f"Graph Query Answer: {query_response['completion']}")
else:
print(f"Graph query failed: {query_response['error']}")
else:
print(f"Entity relationship extraction failed: {entity_graph_response.get('error', 'Unknown error')}")
await client.close()
# if __name__ == "__main__": asyncio.run(entity_graph_example())
文档分块
import asyncio
from mcp.client import Client
async def document_chunking_example():
client = Client("http://localhost:8013")
large_document = """
This is the first paragraph of a potentially very long document. It discusses various concepts.
The second paragraph continues the discussion, adding more details and nuances. Proper chunking
is crucial for processing large texts with Large Language Models, especially those with limited
context windows. Different strategies exist, such as fixed token size, sentence splitting,
or more advanced semantic chunking that tries to keep related ideas together. Overlap between
chunks helps maintain context across boundaries. This paragraph is intentionally made longer
to demonstrate how chunking might split it. It keeps going and going, describing the benefits
of effective text splitting for downstream tasks like summarization, question answering, and
retrieval-augmented generation (RAG). The goal is to create manageable pieces of text that
still retain coherence. Semantic chunking often uses embedding models to find natural breakpoints
in the text's meaning, potentially leading to better results than simple fixed-size chunks.
The final sentence of this example paragraph.
""" * 5 # Make it a bit longer for demonstration
print("Requesting document chunking...")
# Request chunking using a specific method and size
chunking_response = await client.tools.chunk_document(
document=large_document,
chunk_size=100, # Target size in tokens (approximate)
overlap=20, # Token overlap between consecutive chunks
method="semantic" # Options: "token", "sentence", "semantic", "structural" (if available)
)
if chunking_response["success"]:
print(f"Document successfully divided into {chunking_response['chunk_count']} chunks.")
print(f"Method Used: {chunking_response.get('method_used', 'N/A')}") # Confirm method if returned
print("\n--- Example Chunks ---")
for i, chunk in enumerate(chunking_response['chunks'][:3]): # Show first 3 chunks
print(f"Chunk {i+1} (Length: {len(chunk)} chars):")
print(f"'{chunk}'\n")
if chunking_response['chunk_count'] > 3: print("...")
print("----------------------")
# These chunks can now be passed individually to other tools (e.g., summarize_document)
else:
print(f"Document chunking failed: {chunking_response['error']}")
await client.close()
# if __name__ == "__main__": asyncio.run(document_chunking_example())
多提供商补全（与前面的示例重复，为保持结构而保留）
import asyncio
from mcp.client import Client
async def multi_provider_completion_example():
client = Client("http://localhost:8013")
prompt = "What are the main benefits of using the Model Context Protocol (MCP)?"
print(f"Requesting completions for prompt: '{prompt}' from multiple providers...")
multi_response = await client.tools.multi_completion(
prompt=prompt,
providers=[
{"provider": "openai", "model": "gpt-4.1-mini"},
{"provider": "anthropic", "model": "claude-3-5-haiku-20241022"},
{"provider": "gemini", "model": "gemini-2.0-flash-lite"}
# Add more configured providers as needed
],
temperature=0.5,
max_tokens=250
)
if multi_response["success"]:
print("\n--- Multi-completion Results ---")
total_cost = multi_response.get("total_cost", 0.0)
print(f"Total Estimated Cost: ${total_cost:.6f}\n")
for provider_key, result in multi_response["results"].items():
print(f"--- Provider: {provider_key} ---")
if result["success"]:
print(f" Model: {result.get('model', 'N/A')}")
print(f" Cost: ${result.get('cost', 0.0):.6f}")
print(f" Completion:\n{result['completion']}\n")
else:
print(f" Error: {result['error']}\n")
print("------------------------------")
else:
print(f"\nMulti-completion request failed: {multi_response['error']}")
await client.close()
# if __name__ == "__main__": asyncio.run(multi_provider_completion_example())
结构化数据提取（JSON）
import asyncio
from mcp.client import Client
import json
async def json_extraction_example():
client = Client("http://localhost:8013")
text_with_data = """
Meeting Minutes - Project Phoenix - 2024-07-21
Attendees: Alice (Lead), Bob (Dev), Charlie (QA)
Date: July 21, 2024
Project ID: PX-001
Discussion Points:
- Reviewed user feedback from v1.1 testing. Mostly positive.
- Identified performance bottleneck in data processing module. Bob to investigate. Assigned High priority.
- QA cycle for v1.2 planned to start next Monday (2024-07-29). Charlie confirmed readiness.
Action Items:
1. Bob: Investigate performance issue. Due: 2024-07-26. Priority: High. Status: Open.
2. Alice: Prepare v1.2 release notes. Due: 2024-07-28. Priority: Medium. Status: Open.
"""
# Define the desired JSON structure (schema)
desired_schema = {
"type": "object",
"properties": {
"project_name": {"type": "string", "description": "Name of the project"},
"meeting_date": {"type": "string", "format": "date", "description": "Date of the meeting"},
"attendees": {"type": "array", "items": {"type": "string"}, "description": "List of attendee names"},
"action_items": {
"type": "array",
"items": {
"type": "object",
"properties": {
"task": {"type": "string"},
"assigned_to": {"type": "string"},
"due_date": {"type": "string", "format": "date"},
"priority": {"type": "string", "enum": ["Low", "Medium", "High"]},
"status": {"type": "string", "enum": ["Open", "In Progress", "Done"]}
},
"required": ["task", "assigned_to", "due_date", "priority", "status"]
}
}
},
"required": ["project_name", "meeting_date", "attendees", "action_items"]
}
print("Requesting JSON extraction based on schema...")
# Request extraction using a model capable of following JSON schema instructions
json_response = await client.tools.extract_json(
document=text_with_data,
json_schema=desired_schema,
provider="openai", # OpenAI models are generally good at this
model="gpt-4o", # Use a capable model like GPT-4o or Claude 3.5 Sonnet
# provider_preference="quality" # Could also use preference
)
if json_response["success"]:
print("JSON extraction successful.")
print(f"Estimated Cost: ${json_response.get('cost', 0.0):.6f}")
# The extracted data should conform to the schema
extracted_json_data = json_response.get('json_data')
print("\n--- Extracted JSON Data ---")
# Pretty print the JSON
print(json.dumps(extracted_json_data, indent=2))
print("---------------------------")
# Optionally, validate the output against the schema client-side (requires jsonschema library)
# try:
# from jsonschema import validate
# validate(instance=extracted_json_data, schema=desired_schema)
# print("\nClient-side validation successful: Output matches schema.")
# except ImportError:
# print("\n(Install jsonschema to perform client-side validation)")
# except Exception as e:
# print(f"\nClient-side validation failed: {e}")
else:
print(f"JSON Extraction Error: {json_response.get('error', 'Unknown error')}")
if 'details' in json_response: print(f"Details: {json_response['details']}")
await client.close()
# if __name__ == "__main__": asyncio.run(json_extraction_example())
检索增强生成（RAG）查询
import asyncio
from mcp.client import Client
async def rag_query_example():
# This example assumes the Ultimate MCP Server has been configured with a RAG pipeline,
# including a vector store/index containing relevant documents.
client = Client("http://localhost:8013")
query = "What are the latest treatment options for mitigating Alzheimer's disease according to recent studies?"
print(f"Performing RAG query: '{query}'...")
# Call the RAG tool, which handles retrieval and generation
rag_response = await client.tools.rag_query( # Assuming the tool name is 'rag_query'
query=query,
# Optional parameters to control the RAG process:
index_name="medical_research_papers", # Specify the index/collection to search
top_k=3, # Retrieve top 3 most relevant documents/chunks
# filter={"year": {"$gte": 2023}}, # Example filter (syntax depends on vector store)
# generation_model={"provider": "anthropic", "model": "claude-3-5-sonnet-20241022"}, # Specify generation model
# instruction_prompt="Based on the provided context, answer the user's query concisely." # Customize generation prompt
)
if rag_response["success"]:
print("\nRAG query successful.")
print(f"Estimated Cost: ${rag_response.get('cost', 0.0):.6f}") # Includes retrieval + generation cost
print("\n--- Generated Answer ---")
print(rag_response.get('answer', 'No answer generated.'))
print("------------------------")
# The response might also include details about the retrieved sources
retrieved_sources = rag_response.get('sources', [])
if retrieved_sources:
print("\n--- Retrieved Sources ---")
for i, source in enumerate(retrieved_sources):
print(f"Source {i+1}:")
print(f" ID: {source.get('id', 'N/A')}")
print(f" Score: {source.get('score', 'N/A'):.4f}")
# Depending on RAG setup, might include metadata or text snippet
print(f" Content Snippet: {source.get('text', '')[:150]}...")
print("-" * 15)
print("-----------------------")
else:
print("\nNo sources information provided in the response.")
else:
print(f"\nRAG Query Error: {rag_response.get('error', 'Unknown error')}")
if 'details' in rag_response: print(f"Details: {rag_response['details']}")
await client.close()
# if __name__ == "__main__": asyncio.run(rag_query_example())
融合搜索（关键词 + 语义）
import asyncio
from mcp.client import Client
async def fused_search_example():
# This example assumes the server is configured with a hybrid search provider like Marqo.
client = Client("http://localhost:8013")
query = "impact of AI on software development productivity and code quality"
print(f"Performing fused search for: '{query}'...")
# Call the fused search tool
fused_search_response = await client.tools.fused_search( # Assuming tool name is 'fused_search'
query=query,
# --- Parameters specific to the hybrid search backend (e.g., Marqo) ---
index_name="tech_articles_index", # Specify the target index
searchable_attributes=["title", "content"], # Fields to search within
limit=5, # Number of results to return
# Tunable weights for keyword vs. semantic relevance (example)
hybrid_factors={"keyword_weight": 0.4, "semantic_weight": 0.6},
# Optional filter string (syntax depends on backend)
filter_string="publication_year >= 2023 AND source_type='journal'"
# --------------------------------------------------------------------
)
if fused_search_response["success"]:
print("\nFused search successful.")
results = fused_search_response.get("results", [])
print(f"Found {len(results)} hits.")
if results:
print("\n--- Search Results ---")
for i, hit in enumerate(results):
print(f"Result {i+1}:")
# Fields depend on Marqo index structure and what's returned
print(f" ID: {hit.get('_id', 'N/A')}")
print(f" Score: {hit.get('_score', 'N/A'):.4f}") # Combined score
print(f" Title: {hit.get('title', 'N/A')}")
print(f" Content Snippet: {hit.get('content', '')[:150]}...")
# Print highlight info if available
highlights = hit.get('_highlights', {})
if highlights: print(f" Highlights: {highlights}")
print("-" * 15)
print("--------------------")
else:
print("No results found matching the criteria.")
else:
print(f"\nFused Search Error: {fused_search_response.get('error', 'Unknown error')}")
if 'details' in fused_search_response: print(f"Details: {fused_search_response['details']}")
await client.close()
# if __name__ == "__main__": asyncio.run(fused_search_example())
本地文本处理
import asyncio
from mcp.client import Client
async def local_text_processing_example():
client = Client("http://localhost:8013")
# Example assumes a tool named 'process_local_text' exists on the server
# that bundles various non-LLM text operations.
raw_text = " This text has EXTRA whitespace,\n\nmultiple newlines, \t tabs, and needs Case Normalization. "
print("Requesting local text processing operations...")
local_process_response = await client.tools.process_local_text(
text=raw_text,
operations=[
{"action": "trim_whitespace"}, # Remove leading/trailing whitespace
{"action": "normalize_whitespace"}, # Collapse multiple spaces/tabs to single space
{"action": "remove_blank_lines"}, # Remove empty lines
{"action": "lowercase"} # Convert to lowercase
# Other potential actions: uppercase, remove_punctuation, normalize_newlines, etc.
]
)
if local_process_response["success"]:
print("\nLocal text processing successful.")
print(f"Original Text:\n'{raw_text}'")
print(f"\nProcessed Text:\n'{local_process_response['processed_text']}'")
# Note: This operation should ideally have zero LLM cost.
print(f"Cost: ${local_process_response.get('cost', 0.0):.6f}")
else:
print(f"\nLocal Text Processing Error: {local_process_response['error']}")
await client.close()
# if __name__ == "__main__": asyncio.run(local_text_processing_example())
浏览器自动化示例：入门与基本交互
import asyncio
from mcp.client import Client
async def browser_basic_interaction_example():
# This example shows fundamental browser actions controlled by an agent
client = Client("http://localhost:8013")
print("--- Browser Automation: Basic Interaction ---")
# 1. Initialize the browser (creates a browser instance on the server)
print("\nInitializing browser (headless)...")
# `headless=True` runs without a visible GUI window (common for automation)
init_response = await client.tools.browser_init(headless=True, browser_type="chromium")
if not init_response["success"]:
print(f"Browser initialization failed: {init_response.get('error', 'Unknown error')}")
await client.close()
return
print("Browser initialized successfully.")
# Might return session ID if needed for subsequent calls, depends on tool design
# 2. Navigate to a page
target_url = "https://example.com/"
print(f"\nNavigating to {target_url}...")
# `wait_until` controls when navigation is considered complete
nav_response = await client.tools.browser_navigate(
url=target_url,
wait_until="domcontentloaded" # Options: load, domcontentloaded, networkidle, commit
)
if nav_response["success"]:
print(f"Navigation successful.")
print(f" Current URL: {nav_response.get('url', 'N/A')}")
print(f" Page Title: {nav_response.get('title', 'N/A')}")
# The 'snapshot' gives the agent context about the page state (accessibility tree)
# print(f" Snapshot: {nav_response.get('snapshot', 'N/A')}")
else:
print(f"Navigation failed: {nav_response.get('error', 'Unknown error')}")
# Attempt to close browser even if navigation failed
await client.tools.browser_close()
await client.close()
return
# 3. Extract text content using a CSS selector
selector = "h1" # CSS selector for the main heading
print(f"\nExtracting text from selector '{selector}'...")
text_response = await client.tools.browser_get_text(selector=selector)
if text_response["success"]:
print(f"Extracted text: '{text_response.get('text', 'N/A')}'")
else:
print(f"Text extraction failed: {text_response.get('error', 'Unknown error')}")
# Optionally check text_response['snapshot'] for page state at time of failure
# 4. Take a screenshot (optional)
print("\nTaking a screenshot...")
screenshot_response = await client.tools.browser_screenshot(
file_path="example_com_screenshot.png", # Path where server saves the file
full_page=False, # Capture only the viewport
image_format="png" # png or jpeg
)
if screenshot_response["success"]:
print(f"Screenshot saved successfully on server at: {screenshot_response.get('saved_path', 'N/A')}")
# Agent might use this path with a filesystem tool to retrieve the image if needed
else:
print(f"Screenshot failed: {screenshot_response.get('error', 'Unknown error')}")
# 5. Close the browser session
print("\nClosing the browser...")
close_response = await client.tools.browser_close()
if close_response["success"]:
print("Browser closed successfully.")
else:
# Log error, but might happen if browser already crashed
print(f"Browser close failed (might be expected if previous steps failed): {close_response.get('error', 'Unknown error')}")
print("--- Browser Automation Example Complete ---")
await client.close()
# if __name__ == "__main__": asyncio.run(browser_basic_interaction_example())
运行模型锦标赛
import asyncio
from mcp.client import Client
import json
async def model_tournament_example():
client = Client("http://localhost:8013")
# Define the task and prompt for the tournament
task_prompt = "Write a Python function that takes a list of integers and returns a new list containing only the even numbers."
# Optional: Provide ground truth for automated evaluation if the tool supports it
ground_truth_code = """
def get_even_numbers(numbers):
\"\"\"Returns a new list containing only the even numbers from the input list.\"\"\"
return [num for num in numbers if num % 2 == 0]
"""
print("Setting up and running a model tournament for code generation...")
# Call the tournament tool
tournament_response = await client.tools.run_model_tournament(
task_type="code_generation", # Helps select appropriate evaluation metrics
prompt=task_prompt,
# List of models/providers to compete
competitors=[
{"provider": "openai", "model": "gpt-4.1-mini", "temperature": 0.2},
{"provider": "anthropic", "model": "claude-3-5-sonnet-20241022", "temperature": 0.2},
{"provider": "deepseek", "model": "deepseek-coder", "temperature": 0.2}, # Specialized coder model
{"provider": "gemini", "model": "gemini-2.0-pro", "temperature": 0.2},
],
# Criteria for evaluating the generated code
evaluation_criteria=["correctness", "efficiency", "readability", "docstring_quality"],
# Provide ground truth if available for automated correctness checks
ground_truth=ground_truth_code,
# Optional: Specify an LLM to act as the judge for qualitative criteria
evaluation_model={"provider": "anthropic", "model": "claude-3-5-opus-20240229"}, # Use a powerful model for judging
num_rounds=1 # Run multiple rounds for stability if needed
)
if tournament_response["success"]:
print("\n--- Model Tournament Results ---")
print(f"Task Prompt: {task_prompt}")
print(f"Total Estimated Cost: ${tournament_response.get('total_cost', 0.0):.6f}\n")
# Display the ranking
ranking = tournament_response.get("ranking", [])
if ranking:
print("Overall Ranking:")
for i, result in enumerate(ranking):
provider = result.get('provider', 'N/A')
model = result.get('model', 'N/A')
score = result.get('overall_score', 'N/A')
cost = result.get('cost', 0.0)
print(f" {i+1}. {provider}/{model} - Score: {score:.2f}/10 - Cost: ${cost:.6f}")
else:
print("No ranking information available.")
# Display detailed results for each competitor
detailed_results = tournament_response.get("results", {})
if detailed_results:
print("\nDetailed Scores per Competitor:")
for competitor_key, details in detailed_results.items():
print(f" Competitor: {competitor_key}")
print(f" Generated Code:\n```python\n{details.get('output', 'N/A')}\n```")
scores = details.get('scores', {})
if scores:
for criterion, score_value in scores.items():
print(f" - {criterion}: {score_value}")
print("-" * 10)
print("------------------------------")
else:
print(f"\nModel Tournament Failed: {tournament_response.get('error', 'Unknown error')}")
if 'details' in tournament_response: print(f"Details: {tournament_response['details']}")
await client.close()
# if __name__ == "__main__": asyncio.run(model_tournament_example())
用于发现工具的元工具
import asyncio
from mcp.client import Client
import json
async def meta_tools_example():
client = Client("http://localhost:8013")
print("--- Meta Tools Example ---")
# 1. List all available tools
print("\nFetching list of available tools...")
# Assumes a tool named 'list_tools' provides this info
list_tools_response = await client.tools.list_tools(include_schemas=False) # Set True for full schemas
if list_tools_response["success"]:
tools = list_tools_response.get("tools", {})
print(f"Found {len(tools)} available tools:")
for tool_name, tool_info in tools.items():
description = tool_info.get('description', 'No description available.')
print(f" - {tool_name}: {description[:100]}...") # Print truncated description
else:
print(f"Failed to list tools: {list_tools_response.get('error', 'Unknown error')}")
# 2. Get detailed information about a specific tool
tool_to_inspect = "extract_json"
print(f"\nFetching details for tool: '{tool_to_inspect}'...")
# Assumes a tool like 'get_tool_info' or using list_tools with specific name/schema flag
tool_info_response = await client.tools.list_tools(tool_names=[tool_to_inspect], include_schemas=True)
if tool_info_response["success"] and tool_to_inspect in tool_info_response.get("tools", {}):
tool_details = tool_info_response["tools"][tool_to_inspect]
print(f"\nDetails for '{tool_to_inspect}':")
print(f" Description: {tool_details.get('description', 'N/A')}")
# Print the parameter schema if available
schema = tool_details.get('parameters', {}).get('json_schema', {})
if schema:
print(f" Parameter Schema:\n{json.dumps(schema, indent=2)}")
else:
print(" Parameter Schema: Not available.")
else:
print(f"Failed to get info for tool '{tool_to_inspect}': {tool_info_response.get('error', 'Not found or error')}")
# 3. Get tool recommendations for a task (if such a meta tool exists)
task_description = "Read data from a PDF file, extract tables, and save them as CSV."
print(f"\nGetting tool recommendations for task: '{task_description}'...")
# Assumes a tool like 'get_tool_recommendations'
recommendations_response = await client.tools.get_tool_recommendations(
task=task_description,
constraints={"priority": "accuracy", "max_cost_per_doc": 0.10} # Example constraints
)
if recommendations_response["success"]:
print("Recommended Tool Workflow:")
recommendations = recommendations_response.get("recommendations", [])
if recommendations:
for i, step in enumerate(recommendations):
print(f" Step {i+1}: Tool='{step.get('tool', 'N/A')}' - Reason: {step.get('reason', 'N/A')}")
else:
print(" No recommendations provided.")
else:
print(f"Failed to get recommendations: {recommendations_response.get('error', 'Unknown error')}")
print("\n--- Meta Tools Example Complete ---")
await client.close()
# if __name__ == "__main__": asyncio.run(meta_tools_example())
本地命令行文本处理（例如 jq）
import asyncio
from mcp.client import Client
import json
async def local_cli_tool_example():
client = Client("http://localhost:8013")
print("--- Local CLI Tool Example (jq) ---")
# Example JSON data to be processed by jq
json_input_data = json.dumps({
"users": [
{"id": 1, "name": "Alice", "email": "[email protected]", "status": "active"},
{"id": 2, "name": "Bob", "email": "[email protected]", "status": "inactive"},
{"id": 3, "name": "Charlie", "email": "[email protected]", "status": "active"}
],
"metadata": {"timestamp": "2024-07-21T12:00:00Z"}
})
# Define the jq filter to apply
# This filter selects active users and outputs their name and email
jq_filter = '.users[] | select(.status=="active") | {name: .name, email: .email}'
print(f"\nRunning jq with filter: '{jq_filter}' on input JSON...")
# Call the server tool that wraps jq (e.g., 'run_jq')
jq_result = await client.tools.run_jq(
args_str=jq_filter, # Pass the filter as arguments (check tool spec how it expects filters)
input_data=json_input_data, # Provide the JSON string as input
# Additional options might be available depending on the tool wrapper:
# e.g., output_format="json_lines" or "compact_json"
)
if jq_result["success"]:
print("jq execution successful.")
# stdout typically contains the result of the jq filter
print("\n--- jq Output (stdout) ---")
print(jq_result.get("stdout", "No output"))
print("--------------------------")
# stderr might contain warnings or errors from jq itself
stderr_output = jq_result.get("stderr")
if stderr_output:
print("\n--- jq Stderr ---")
print(stderr_output)
print("-----------------")
# This should have minimal or zero cost as it runs locally on the server
print(f"\nCost: ${jq_result.get('cost', 0.0):.6f}")
else:
print(f"\njq Execution Error: {jq_result.get('error', 'Unknown error')}")
print(f"Stderr: {jq_result.get('stderr', 'N/A')}")
print("\n--- Local CLI Tool Example Complete ---")
await client.close()
# if __name__ == "__main__": asyncio.run(local_cli_tool_example())
动态 API 集成
import asyncio
from mcp.client import Client
import json
async def dynamic_api_example():
# This example assumes the server has tools like 'register_api', 'list_registered_apis',
# 'call_dynamic_tool', and 'unregister_api'.
client = Client("http://localhost:8013")
print("--- Dynamic API Integration Example ---")
# 1. Register an external API using its OpenAPI (Swagger) specification URL
api_name_to_register = "public_cat_facts"
openapi_spec_url = "https://catfact.ninja/docs/api-docs.json" # Example public API spec
print(f"\nRegistering API '{api_name_to_register}' from {openapi_spec_url}...")
register_response = await client.tools.register_api(
api_name=api_name_to_register,
openapi_url=openapi_spec_url,
# Optional: Provide authentication details if needed (e.g., Bearer token, API Key)
# authentication={"type": "bearer", "token": "your_api_token"},
# Optional: Set default headers
# default_headers={"X-Custom-Header": "value"},
# Optional: Cache settings for API responses (if tool supports it)
cache_ttl=300 # Cache responses for 5 minutes
)
if register_response["success"]:
print(f"API '{api_name_to_register}' registered successfully.")
print(f" Registered {register_response.get('tools_count', 0)} new MCP tools derived from the API.")
print(f" Tools Registered: {register_response.get('tools_registered', [])}")
else:
print(f"API registration failed: {register_response.get('error', 'Unknown error')}")
await client.close()
return
# 2. List currently registered dynamic APIs
print("\nListing registered dynamic APIs...")
list_apis_response = await client.tools.list_registered_apis()
if list_apis_response["success"]:
registered_apis = list_apis_response.get("apis", {})
print(f"Currently registered APIs: {list(registered_apis.keys())}")
# print(json.dumps(registered_apis, indent=2)) # Print full details
else:
print(f"Failed to list registered APIs: {list_apis_response.get('error', 'Unknown error')}")
# 3. Call a dynamically created tool corresponding to an API endpoint
# The tool name is typically derived from the API name and endpoint's operationId or path.
# Check the 'tools_registered' list from step 1 or documentation for the exact name.
# Let's assume the tool for GET /fact is 'public_cat_facts_getFact'
dynamic_tool_name = "public_cat_facts_getFact" # Adjust based on actual registered name
print(f"\nCalling dynamic tool '{dynamic_tool_name}'...")
call_response = await client.tools.call_dynamic_tool(
tool_name=dynamic_tool_name,
# Provide inputs matching the API endpoint's parameters
inputs={
# Example query parameter for GET /fact (check API spec)
"max_length": 100
}
)
if call_response["success"]:
print("Dynamic tool call successful.")
# The result usually contains the API's response body and status code
print(f" Status Code: {call_response.get('status_code', 'N/A')}")
print(f" Response Body:\n{json.dumps(call_response.get('response_body', {}), indent=2)}")
else:
print(f"Dynamic tool call failed: {call_response.get('error', 'Unknown error')}")
print(f" Status Code: {call_response.get('status_code', 'N/A')}")
print(f" Response Body: {call_response.get('response_body', 'N/A')}")
# 4. Unregister the API when no longer needed (optional cleanup)
print(f"\nUnregistering API '{api_name_to_register}'...")
unregister_response = await client.tools.unregister_api(api_name=api_name_to_register)
if unregister_response["success"]:
print(f"API unregistered successfully. Removed {unregister_response.get('tools_count', 0)} tools.")
else:
print(f"API unregistration failed: {unregister_response.get('error', 'Unknown error')}")
print("\n--- Dynamic API Integration Example Complete ---")
await client.close()
# if __name__ == "__main__": asyncio.run(dynamic_api_example())
OCR 使用示例
import asyncio
from mcp.client import Client
import os
async def ocr_example():
# Requires 'ocr' extras installed: uv pip install -e ".[ocr]"
# Also requires Tesseract OCR engine installed on the server host system.
client = Client("http://localhost:8013")
print("--- OCR Tool Example ---")
# --- Create dummy files for testing ---
# In a real scenario, these files would exist on a path accessible by the server.
# Ensure the server process has permissions to read these files.
dummy_files_dir = "ocr_test_files"
os.makedirs(dummy_files_dir, exist_ok=True)
dummy_pdf_path = os.path.join(dummy_files_dir, "dummy_document.pdf")
dummy_image_path = os.path.join(dummy_files_dir, "dummy_image.png")
# Create a simple dummy PDF (requires reportlab - pip install reportlab)
try:
from reportlab.pdfgen import canvas
from reportlab.lib.pagesizes import letter
c = canvas.Canvas(dummy_pdf_path, pagesize=letter)
c.drawString(100, 750, "This is page 1 of a dummy PDF.")
c.drawString(100, 730, "It contains some text for OCR testing.")
c.showPage()
c.drawString(100, 750, "This is page 2.")
c.save()
print(f"Created dummy PDF: {dummy_pdf_path}")
except ImportError:
print("Could not create dummy PDF: reportlab not installed. Skipping PDF test.")
dummy_pdf_path = None
except Exception as e:
print(f"Error creating dummy PDF: {e}. Skipping PDF test.")
dummy_pdf_path = None
# Create a simple dummy PNG image (requires Pillow - pip install Pillow)
try:
from PIL import Image, ImageDraw, ImageFont
img = Image.new('RGB', (400, 100), color = (255, 255, 255))
d = ImageDraw.Draw(img)
# Use a default font if possible, otherwise basic text
try: font = ImageFont.truetype("arial.ttf", 15)
except IOError: font = ImageFont.load_default()
d.text((10,10), "Dummy Image Text for OCR\nLine 2 of text.", fill=(0,0,0), font=font)
img.save(dummy_image_path)
print(f"Created dummy Image: {dummy_image_path}")
except ImportError:
print("Could not create dummy Image: Pillow not installed. Skipping Image test.")
dummy_image_path = None
except Exception as e:
print(f"Error creating dummy Image: {e}. Skipping Image test.")
dummy_image_path = None
# --- End of dummy file creation ---
# 1. Extract text from the PDF using OCR and LLM correction
if dummy_pdf_path:
print(f"\nExtracting text from PDF: {dummy_pdf_path} (using hybrid method)...")
pdf_text_result = await client.tools.extract_text_from_pdf(
file_path=dummy_pdf_path, # Server needs access to this path
extraction_method="hybrid", # Try direct extraction, fallback to OCR
max_pages=2, # Limit pages to process
reformat_as_markdown=True, # Request markdown formatting
# Optional: Use an LLM to correct/improve the raw OCR text
llm_correction_model={"provider": "openai", "model": "gpt-4.1-mini"}
)
if pdf_text_result["success"]:
print("PDF text extraction successful.")
print(f" Method Used: {pdf_text_result.get('extraction_method_used', 'N/A')}")
print(f" Cost (incl. LLM correction): ${pdf_text_result.get('cost', 0.0):.6f}")
print("\n--- Extracted PDF Text (Markdown) ---")
print(pdf_text_result.get("text", "No text extracted."))
print("-------------------------------------")
else:
print(f"PDF OCR failed: {pdf_text_result.get('error', 'Unknown error')}")
if 'details' in pdf_text_result: print(f"Details: {pdf_text_result['details']}")
else:
print("\nSkipping PDF OCR test as dummy file could not be created.")
# 2. Process the image file with OCR and preprocessing
if dummy_image_path:
print(f"\nProcessing image OCR: {dummy_image_path} with preprocessing...")
image_text_result = await client.tools.process_image_ocr(
image_path=dummy_image_path, # Server needs access to this path
# Optional preprocessing steps (require OpenCV on server)
preprocessing_options={
"grayscale": True,
# "threshold": "otsu", # e.g., otsu, adaptive
# "denoise": True,
# "deskew": True
},
ocr_language="eng" # Specify language(s) for Tesseract e.g., "eng+fra"
# Optional LLM enhancement for image OCR results
# llm_enhancement_model={"provider": "gemini", "model": "gemini-2.0-flash-lite"}
)
if image_text_result["success"]:
print("Image OCR successful.")
print(f" Cost (incl. LLM enhancement): ${image_text_result.get('cost', 0.0):.6f}")
print("\n--- Extracted Image Text ---")
print(image_text_result.get("text", "No text extracted."))
print("----------------------------")
else:
print(f"Image OCR failed: {image_text_result.get('error', 'Unknown error')}")
if 'details' in image_text_result: print(f"Details: {image_text_result['details']}")
else:
print("\nSkipping Image OCR test as dummy file could not be created.")
# --- Clean up dummy files ---
# try:
# if dummy_pdf_path and os.path.exists(dummy_pdf_path): os.remove(dummy_pdf_path)
# if dummy_image_path and os.path.exists(dummy_image_path): os.remove(dummy_image_path)
# if os.path.exists(dummy_files_dir): os.rmdir(dummy_files_dir) # Only if empty
# except Exception as e:
# print(f"\nError cleaning up dummy files: {e}")
# --- End cleanup ---
print("\n--- OCR Tool Example Complete ---")
await client.close()
# if __name__ == "__main__": asyncio.run(ocr_example())
（注意：许多涉及文件路径的示例都假设服务进程可以访问这些路径。对于 Docker 部署，通常需要进行卷映射。）
✨ 自主文档优化器
Ultimate MCP Server 包含一项强大功能，可自主分析、测试和优化已注册 MCP 工具的文档。此功能由 ultimate/tools/docstring_refiner.py 实现，有助于提升 Claude 等大语言模型（LLM）调用工具时的易用性和可靠性。
工作原理
文档优化器采用系统化的迭代方法：
- Agent 模拟：模拟大模型 Agent 如何解读当前文档（文档字符串、架构、示例），从而识别可能存在的歧义或正确调用所需的信息缺失。
- 自适应测试生成：根据工具的输入架构（参数类型、约束、必填字段）、模拟结果以及之前优化迭代中出现的故障，创建多样化的测试用例，以实现良好的覆盖率。
- 架构感知测试：在执行前，根据工具的架构验证生成的测试输入。在服务端环境中，对实际工具实现执行有效的测试。
- 集成故障分析：如果测试失败（例如，输出错误或抛出异常），多个大模型会结合该次测试运行所用的特定版本文档来分析故障，从而找出文档中的不足。
- 结构化改进建议：系统根据分析结果生成具体、有针对性的改进建议：
- 描述： 改写内容或补充说明。
- 架构： 通过 JSON Patch 操作提出更改建议（例如，为参数添加描述、优化类型、添加示例）。
- 用法示例： 生成新示例或优化现有示例。
- 经过验证的架构修补：在内存中将建议的 JSON Patch 应用于架构，并在接受更改供下一轮迭代使用前，验证生成的架构结构。
- 迭代优化：重复这一循环（生成测试 → 执行 → 分析故障 → 提出改进建议 → 修补架构），直到测试稳定通过或达到最大迭代次数。
- 可选精简：迭代结束后，进行最后一轮精简，使文档更简洁，同时确保保留测试中发现的关键信息。
优势
- 减少人工工作量：自动完成为大模型编写和维护高质量工具文档这一往往繁琐的工作。
- 提升 Agent 性能：生成更清晰、更准确的文档，减少大模型尝试使用工具时出错的情况。
- 发现边缘情况：测试过程能够发现人工撰写者可能忽略的歧义和边缘情况。
- 提高一致性：有助于为所有工具的文档建立更统一的风格和详略程度。
- 适应反馈：直接从模拟 Agent 的失败中学习，针对具体的文档不足进行改进。
- 架构演进：根据使用模拟，逐步、经过验证地改进工具架构。
- 详细报告：提供整个优化过程的完整日志和报告，包括运行的测试、遇到的故障以及所做的更改。
局限性和注意事项
- 成本和时间：计算成本可能较高，耗时也可能较长，因为每个工具每轮迭代都需要多次调用大模型（用于模拟、测试生成、故障分析和改进建议）。
- 资源消耗：可能需要大量 CPU 和内存，尤其是在优化大量工具或使用大型大模型进行分析时。
- 依赖大模型：优化质量在很大程度上取决于用于分析和生成的大模型的能力。
- 架构复杂性：对于高度复杂或嵌套的架构，大模型可能难以生成正确且有意义的 JSON Patch。
- 确定性：此过程涉及大模型，因此不同运行的结果可能无法完全确定。
- 维护复杂性：优化器本身是一个复杂系统，包含需要维护的依赖项。
适用场景
以下情况尤其适合使用此功能：
- 您向 LLM Agent 暴露了大量 MCP 工具。
- 您发现工具经常调用失败，可能是因为 Agent 误解了文档。
- 您正在积极开发或扩展工具生态，需要确保文档一致且高质量。
- 您希望主动提升 Agent 的可靠性和性能，而不一定要修改底层工具代码。
- 您有预算（LLM 额度）和时间投入这一自动化质量改进流程。
用法示例（服务端调用）
文档优化器通常作为服务端维护或管理任务调用，而不会直接作为 MCP 工具开放给外部 Agent 调用。
# This code snippet shows how the refiner might be called from within the
# server's environment (e.g., via a CLI command or admin interface).
# Assume necessary imports and context setup:
# from ultimate_mcp_server.tools.docstring_refiner import refine_tool_documentation
# from ultimate_mcp_server.core import mcp_context # Represents the server's context
async def invoke_doc_refiner_task():
# Ensure mcp_context is properly initialized with registered tools, config, etc.
print("Starting Autonomous Documentation Refinement Task...")
# Example: Refine documentation for a specific list of tools
refinement_result = await refine_tool_documentation(
tool_names=["extract_json", "browser_navigate", "chunk_document"], # Tools to refine
max_iterations=3, # Limit refinement cycles per tool
refinement_model_config={ # Specify LLM for refinement tasks
"provider": "anthropic",
"model": "claude-3-5-sonnet-20241022"
},
testing_model_config={ # Optional: Specify LLM for test generation/simulation
"provider": "openai",
"model": "gpt-4o"
},
enable_winnowing=True, # Apply final streamlining pass
stop_on_first_error=False, # Continue refining other tools if one fails
ctx=mcp_context # Pass the server's MCP context
)
# Example: Refine all available tools (potentially very long running)
# refinement_result = await refine_tool_documentation(
# refine_all_available=True,
# max_iterations=2,
# ctx=mcp_context
# )
print("\nDocumentation Refinement Task Complete.")
# Process the results
if refinement_result["success"]:
print(f"Successfully processed {len(refinement_result.get('refined_tools', []))} tools.")
# The actual docstrings/schemas of the tools in mcp_context might be updated in-memory.
# Persisting these changes would require additional logic (e.g., writing back to source files).
print("Detailed report available in the result object.")
# print(refinement_result.get('report')) # Contains detailed logs and changes
else:
print(f"Refinement task encountered errors: {refinement_result.get('error', 'Unknown error')}")
# Check the report for details on which tools failed and why.
# To run this, it would need to be integrated into the server's startup sequence,
# a dedicated CLI command, or an administrative task runner.
# e.g., await invoke_doc_refiner_task()
✅ 示例库和测试框架
Ultimate MCP Server 在 examples/ 目录中提供了大量 35+ 个端到端示例。这些示例有双重用途：
- 动态文档：展示几乎所有工具和功能的实用、真实场景用法。
- 集成测试套件：构成全面的测试套件，确保所有组件能够正确协同工作。
示例结构和组织
- 分类整理：示例按功能分组（例如，
model_integration、tool_specific、workflows、advanced_features）。 - 独立运行：每个示例（
*.py）都是可运行的 Python 脚本，使用mcp-client与正在运行的服务实例交互。 - 输出清晰：使用
Rich库生成格式化的彩色控制台输出，清楚展示请求、响应、成本、耗时和结果。 - 错误处理：示例包含基本的错误检查，确保演示过程稳健。
丰富的可视化输出
您可以看到内容丰富的控制台输出，包括：
- 📊 汇总结果和统计信息的表格。
- 🎨 代码和 JSON 的语法高亮。
- ⏳ 进度指示器或详细的步骤日志。
- 🖼️ 用于组织输出内容的面板。
输出示例片段：
╭────────────────────── Tournament Results ───────────────────────╮
│ [1] claude-3-5-haiku-20241022: Score 8.7/10 │
│ Cost: $0.00013 │
│ ... │
╰────────────────────────────────────────────────────────────────╯
自定义与学习
- 灵活调整：可轻松修改示例，以使用您的 API 密钥（通过
.env）、不同的模型、自定义提示词或输入文件。 - 命令行参数：许多示例支持用于自定义的参数（例如，
--model、--input-file、--headless）。 - 便于学习：了解 AI 应用结构、工具选择、参数调整、错误处理、成本优化和集成模式的最佳实践。
全面的测试框架
run_all_demo_scripts_and_check_for_errors.py 脚本会协调运行所有示例，并将其作为测试套件：
- 自动执行：按顺序发现并运行
examples/*.py。 - 验证：检查退出代码和
stderr，并根据预定义模式区分实际错误与预期消息（例如，缺少 API 密钥的警告）。 - 报告：生成通过、失败和跳过的测试摘要报告，并附上详细日志。
测试框架配置示例片段：
"sql_database_interactions_demo.py": {
"expected_exit_code": 0,
"allowed_stderr_patterns": [
r"Could not compute statistics...", # Known non-fatal warning
r"Connection failed...", # Expected if DB not set up
r"Configuration not yet loaded..." # Standard info message
]
}
运行示例套件
# Ensure the Ultimate MCP Server is running in a separate terminal
# Run the entire test suite
python run_all_demo_scripts_and_check_for_errors.py
# Run a specific example script directly
python examples/browser_automation_demo.py --headless
# Run an example with custom arguments
python examples/text_redline_demo.py --input-file1 path/to/doc1.txt --input-file2 path/to/doc2.txt
这个示例库与测试框架相结合，为理解、使用和验证 Ultimate MCP Server 的功能提供了宝贵资源。
💻 CLI 命令
Ultimate MCP Server 自带命令行界面（umcp），用于服务管理和工具交互：
# Show available commands and global options
umcp --help
# --- Server Management ---
# Start the server (loads .env, registers tools)
umcp run [--host HOST] [--port PORT] [--include-tools tool1 tool2] [--exclude-tools tool3 tool4]
# --- Information ---
# List configured LLM providers
umcp providers [--check] [--models]
# List available tools
umcp tools [--category CATEGORY] [--examples]
# --- Testing & Interaction ---
# Test connection and basic generation for a specific provider
umcp test <provider_name> [--model MODEL_NAME] [--prompt TEXT]
# Generate a completion directly from the CLI
umcp complete --provider <provider_name> --model <model_name> --prompt "Your prompt here" [--temperature N] [--max-tokens N] [--system TEXT] [--stream]
# --- Cache Management ---
# View or clear the request cache
umcp cache [--status] [--clear]
# --- Benchmark ---
umcp benchmark [--providers P1 P2] [--models M1 M2] [--prompt TEXT] [--runs N]
# --- Examples ---
umcp examples [--list] [<example_name>] [--category CATEGORY]
每个命令通常还有其他选项。使用 umcp COMMAND --help 查看特定命令的选项（例如，umcp complete --help）。
🛠️ 高级配置
配置主要通过环境变量管理，启动时通常从项目根目录下的 .env 文件加载。
服务配置
SERVER_HOST：（默认值：127.0.0.1）要绑定的网络接口。使用0.0.0.0可监听所有接口（Docker 容器或外部访问时必需）。SERVER_PORT：（默认值：8013）服务监听的端口。API_PREFIX：（默认值：/）所有 API 端点的 URL 前缀（例如，设为/mcp/v1可在该路径下提供服务）。WORKERS：（可选，例如4）Web 服务（如 Uvicorn）的工作进程数。请根据 CPU 核心数调整。
工具筛选（启动控制）
使用 CLI 标志控制服务启动时注册哪些工具：
--include-tools tool1,tool2,...：仅注册指定工具。--exclude-tools tool3,tool4,...：注册除指定工具外的所有工具。
这适用于创建轻量级实例、管理依赖项或限制 Agent 的能力。# 示例：仅启用文件系统和基本补全工具 umcp run --include-tools read_file,write_file,list_directory,completion # 示例：启用除浏览器自动化以外的所有工具 umcp run --exclude-tools browser_init,browser_navigate,browser_click
日志配置
LOG_LEVEL：（默认值：INFO）控制日志详细程度（DEBUG、INFO、WARNING、ERROR、CRITICAL）。DEBUG会输出非常详细的日志。USE_RICH_LOGGING：（默认值：true）通过 Rich 库启用彩色结构化控制台日志。设为false可输出纯文本日志（更适合重定向到文件或某些日志系统）。LOG_FORMAT：（可选）指定 Pythonlogging格式字符串，以自定义日志格式（如果USE_RICH_LOGGING=false）。LOG_TO_FILE：（可选，例如/var/log/ultimate_mcp_server.log）日志同时写入的文件路径（除控制台外）。请确保服务进程具有写入权限。
缓存配置
CACHE_ENABLED：（默认值：true）全局启用或禁用响应缓存。CACHE_TTL：（默认值：86400秒 = 24 小时）缓存项的默认生存时间。特定工具可能会覆盖此设置。CACHE_TYPE：（默认值：memory）后端存储。请查看实现以了解支持的类型（例如，memory、redis、diskcache）。diskcache提供持久化存储。CACHE_DIR：（默认值：./.cache）当CACHE_TYPE=diskcache时使用的目录。请确保具有写入权限。CACHE_MAX_SIZE：（可选，例如diskcache中的1000个项目或536870912（512MB））设置缓存大小限制。REDIS_URL：（CACHE_TYPE=redis时必需）Redis 服务的连接 URL（例如，redis://localhost:6379/0）。
提供商超时与重试
PROVIDER_TIMEOUT：（默认值：120）等待大模型提供商 API 响应的默认超时时间（秒）。PROVIDER_MAX_RETRIES：（默认值：3）向提供商发送失败请求时的默认重试次数（适用于限流或临时服务问题等可重试错误）。采用指数退避策略。- 特定提供商可能支持通过专用变量覆盖设置（例如，
OPENAI_TIMEOUT、ANTHROPIC_MAX_RETRIES）。请查看配置加载逻辑或文档。
工具专属配置
各个工具可能会从环境变量加载自己的配置。例如：
ALLOWED_DIRS：文件系统工具被限制访问的基础目录列表，以逗号分隔。这对安全至关重要。PLAYWRIGHT_BROWSER_TYPE：（默认值：chromium）Playwright 工具使用的浏览器（chromium、firefox、webkit）。PLAYWRIGHT_TIMEOUT：Playwright 操作的默认超时时间。DATABASE_URL：SQL 数据库交互工具的连接字符串（使用 SQLAlchemy）。MARQO_URL：融合搜索工具使用的 Marqo 实例 URL。TESSERACT_CMD：如果 Tesseract 可执行文件不在系统标准 PATH 中，则指定其路径（用于 OCR）。
启动服务前，务必确保环境变量已正确设置。更改通常需要重启服务才能生效。
☁️ 部署注意事项
虽然 umcp run 或 docker compose up 适用于开发，但若要实现更可靠的部署，请考虑以下事项：
1. 作为后台服务运行
确保服务持续运行并自动重启。
- **
systemd（Linux）：**创建服务单元文件（.service），以便通过systemctl start|stop|restart|status管理进程。它能提供可靠的控制和日志集成。 - **
supervisor：**一个以 Python 编写的进程控制系统。配置supervisord来监控和管理服务进程。 - **Docker 重启策略：**在
docker run命令或docker-compose.yml中使用--restart unless-stopped或--restart always，让 Docker 管理重启。
2. 使用反向代理（Nginx、Caddy、Apache、Traefik）
强烈建议在 Ultimate MCP Server 前部署反向代理：
- 🔒 **HTTPS/SSL 终止：**处理 SSL 证书（例如，通过 Caddy/Certbot 使用 Let's Encrypt），对外部流量进行加密。
- ⚖️ **负载均衡：**如果运行多个服务实例以实现高可用性或扩展，可分配流量。
- 🗺️ **路径路由：**将简洁的外部 URL（例如
https://api.yourdomain.com/mcp/）映射到内部服务（http://localhost:8013）。如有需要，请配置API_PREFIX。 - 🛡️ **安全标头：**添加
Strict-Transport-Security（HSTS）、Content-Security-Policy（CSP）等重要标头。 - 🚦 **访问控制：**实现 IP 允许列表、基本身份验证，或集成 OAuth2 代理。
- ⏳ **缓冲/缓存：**可提供额外的请求/响应缓冲或缓存层。
- ⏱️ **超时：**独立于应用服务管理连接超时。
Nginx location 块示例（简化版）：
location /mcp/ { # Match your desired public path (corresponds to API_PREFIX if set)
proxy_pass http://127.0.0.1:8013/; # Point to the internal server (note trailing /)
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# Increase timeouts for potentially long-running AI tasks
proxy_connect_timeout 60s;
proxy_send_timeout 300s;
proxy_read_timeout 300s;
# Optional: Add basic authentication
# auth_basic "Restricted Access";
# auth_basic_user_file /etc/nginx/.htpasswd;
}
3. 容器编排（Kubernetes、Docker Swarm）
如需实现可扩展、托管式部署：
- ❤️ **健康检查：**在部署清单中使用服务的
/healthz端点（或类似端点）实现并配置存活和就绪探针。 - 🔑 **配置：**使用 ConfigMaps 和 Secrets（Kubernetes）或 Docker Secrets/Configs 安全管理环境变量和 API 密钥，而不是将其打包到镜像中或仅依赖
.env文件。 - ⚙️ **资源限制：**为容器定义合适的 CPU 和内存请求/限制，确保性能稳定，并避免节点资源耗尽。
- 🌐 **服务发现：**使用编排器内置的服务发现功能，而不是硬编码 IP 或主机名。在内部暴露服务（例如 ClusterIP），并使用 Ingress 控制器提供外部访问。
- 💾 **持久化存储：**如果使用需要持久化的功能（例如
diskcache、持久化记忆、文件存储），请配置持久卷（PVs/PVCs）。
4. 资源分配
- RAM： 确保有足够的内存，尤其是在使用大模型、内存缓存、处理大型文档或运行内存密集型工具（如浏览器自动化或某些数据处理任务）时。请监控内存使用情况。
- CPU： 监控 CPU 负载。LLM 推理本身可能并不受 CPU 限制（通常由 GPU/TPU 负责），但其他工具（OCR、本地处理、处理请求的 Web 服务器）可能会受其限制。请考虑工作进程数量（
WORKERS环境变量）。 - 磁盘 I/O： 如果使用持久化缓存（
diskcache）或执行大量文件系统操作，磁盘 I/O 可能成为瓶颈。必要时请使用高速存储（SSD）。 - 网络： 确保带宽充足，尤其是在处理大型文档、图像或频繁收到大型 API 响应时。
💸 通过委派节省成本
与每项任务都只使用 Claude 3.7 Sonnet 或 GPT-4o 等高端模型相比，使用 Ultimate MCP Server 进行智能委派可以大幅节省成本。
| 任务场景 | 仅使用高端模型（估算） | 通过 MCP 服务委派（估算） | 预计节省 | 备注 |
|---|---|---|---|---|
| 总结 100 页文档 | ~$4.50 - $6.00 | ~$0.45 - $0.70 (Gemini Flash) | ~90% | 分块处理 + 并行生成低成本摘要 |
| 从 50 条记录中提取数据 | ~$2.25 - $3.00 | ~$0.35 - $0.50 (GPT-4.1 Mini) | ~84% | 使用高性价比模型进行批量处理 |
| 生成 20 个内容创意 | ~$0.90 - $1.20 | ~$0.12 - $0.20 (DeepSeek/Haiku) | ~87% | 使用低成本模型完成简单的生成任务 |
| 处理 1,000 条客户查询 | ~$45.00 - $60.00 | ~$7.50 - $12.00 (混合模型) | ~83% | 根据查询复杂度进行路由 |
| 对 10 份扫描件进行 OCR 并提取内容 | ~$1.50 - $2.50 (如果使用 LLM OCR) | ~$0.20 - $0.40 (OCR + LLM 修正) | ~85% | 使用专用 OCR + 低成本 LLM 进行修正 |
| 基础网页抓取与总结 | ~$0.50 - $1.00 | ~$0.10 - $0.20 (Browser + Haiku) | ~80% | 使用 Browser 工具 + 低成本 LLM 进行总结 |
（以上费用仅供参考，基于典型的 token 数量和 2024 年的近似定价。实际费用在很大程度上取决于文档大小、复杂度、所用的具体模型以及服务提供商的当前定价。）
节省成本的方式：
- 根据任务匹配模型： 仅对需要深度推理、创造力或复杂指令遵循能力的任务使用昂贵模型。
- 利用低成本模型： 将总结、提取、简单问答、格式整理等任务委派给成本低得多的模型（如 Gemini Flash、Claude Haiku、GPT-4.1 Mini、DeepSeek Chat）。
- 使用专用工具： 在适当情况下使用非 LLM 工具（Filesystem、OCR、Browser、CLI 工具、Database），避免为这些操作调用 LLM API。
- 缓存： 减少针对相同或语义相似请求的重复 API 调用。
Ultimate MCP Server 充当智能路由层，使复杂的 Agent 架构能够实现这些成本优化。
🧠 AI 间委派为何重要
由 Ultimate MCP Server 等系统促成的 AI 间委派，其战略意义不止于节省成本：
让先进 AI 能力普及化
- 通过将常规工作委派出去，让更广泛的应用能够切实运用前沿推理模型（如 Claude 3.7、GPT-4o）的强大能力。
- 让受预算限制的组织能够将顶级 AI 能力用于关键推理步骤，同时有效控制总体成本。
- 推动行业更高效、更广泛地利用 AI 资源。
经济资源优化
- 体现了 AI 使用中的一项基本经济优化原则：仅在确实需要顶级 LLM 推理的独特价值时，才使用成本最高的资源。
- 将复杂推理、创造力、细致理解和编排任务交由高能力模型处理。
- 将常规数据处理、提取、格式化和较简单的问答交由高性价比模型处理。
- 将非 LLM 的专用任务（网页抓取、文件 I/O、数据库查询）交由专用工具处理，避免不必要的 LLM 调用。
- 整个系统旨在以显著降低的综合成本，实现接近顶级模型的性能和能力。
- 通过智能路由和缓存，将原本可能难以预测的 LLM API 成本转化为更可控的支出。
可持续的 AI 架构
- 通过减少每项任务都使用最大模型所带来的计算需求，推动更可持续地使用 AI。
- 构建与能力相匹配的分层 AI 资源分配方案。
- 让更广泛的实验和开发成为可能，因为许多迭代可以使用成本更低的模型或工具。
- 提供可扩展的 AI 集成方案，使其能随业务需求增长，同时避免成本失控。
技术演进路径
- 体现了 AI 应用架构的一项重要演进：从对单一模型的单体调用，转向分布式、多 Agent、多模型工作流。
- 支持对涉及多种工具和模型的复杂处理管线进行精密的 AI 驱动编排。
- 为 AI 系统奠定基础，使其有望推理自身的资源使用情况并动态优化。
- 推动构建更自主、可自我优化的 AI 系统，使其能够根据上下文、成本和所需质量，智能地决定任务委派方式。
AI 效率的未来
- Ultimate MCP Server 展现了一种未来：AI 系统主动管理并优化自身的运营成本和资源使用。
- 更高能力的模型充当智能编排者或“管理者”，管理由专用工具和更具性价比的“工作”模型组成的生态系统。
- AI 工作流将日益精密，并有望实现自组织且具备韧性。
- 组织可以经济可行且可扩展地利用全方位 AI 能力，涵盖从基础处理到高级推理。
这一高效、智能委派且可自我优化的 AI 系统愿景，代表了实用 AI 部署的下一个前沿方向，超越了当前几乎所有任务都经常使用单一强大（且昂贵）模型的范式。
🧱 架构
MCP 集成的工作原理
Ultimate MCP Server 基于 Model Context Protocol（MCP）原生构建：
- MCP 服务核心：实现一个 Web 服务（例如使用 FastAPI），监听符合 MCP 规范的传入 HTTP 请求（通常是发送到特定端点的 POST 请求）。
- 工具注册：启动期间，服务会发现并注册所有可用的工具实现。每个工具都会提供元数据，包括名称、描述以及输入／输出模式（通常是转换为 JSON Schema 的 Pydantic 模型）。此注册表让服务（以及可能的 Agent）能够了解有哪些工具可用，以及如何使用这些工具。
- 工具调用：当 MCP 客户端（如 Claude 或其他应用程序）发送有效的 MCP 请求，指定工具名称和参数时，服务核心会将请求路由到相应的已注册工具的执行逻辑。
- 上下文传递与执行：工具接收经过验证的输入参数，并执行相应操作（调用大模型、与 Playwright 交互、查询数据库、操作文件等）。
- 结构化响应：工具的执行结果（或错误）会打包成标准的 MCP 响应格式，通常包括状态（成功／失败）、输出数据（符合工具的输出模式）、成本信息，以及可能包含的其他元数据。
- 返回客户端：MCP 服务核心通过 HTTP 将结构化的 MCP 响应发送回发起请求的客户端。
遵循 MCP 标准可确保与任何兼容 MCP 的 Agent 或客户端应用程序无缝、可预测地集成。
组件图
+---------------------+ MCP Request +------------------------------------+ API Request +-----------------+
| MCP Agent/Client | ----------------------> | Ultimate MCP Server | ----------------------> | LLM Providers |
| (e.g., Claude 3.7) | <---------------------- | (FastAPI + MCP Core + Tool Logic) | <---------------------- | (OpenAI, Anthro.)|
+---------------------+ MCP Response +------------------+-----------------+ API Response +--------+--------+
| |
| Tool Invocation | External API Call
▼ ▼
+-----------------------------------------------------------+------------------------------------------------------------+
| Internal Services & Tool Implementations |
| +-------------------+ +-------------------+ +-------------------+ +-------------------+ +-------------------+ |
| | Completion/LLM | | Document Proc. | | Data Extraction | | Browser Automation| | Excel Automation | |
| | (Routing/Provider)| | (Chunking, Sum.) | | (JSON, Table) | | (Playwright) | | (OpenPyXL/COM) | |
| +---------+---------+ +-------------------+ +-------------------+ +-------------------+ +-------------------+ |
| | |
| +---------+---------+ +-------------------+ +-------------------+ +-------------------+ +-------------------+ |
| | Cognitive Memory | | Filesystem Ops | | SQL Database | | Entity/Graph | | Vector/RAG | |
| | (Storage/Query) | | (Secure Access) | | (SQLAlchemy) | | (NetworkX) | | (Vector Stores) | |
| +-------------------+ +-------------------+ +-------------------+ +-------------------+ +---------+---------+ |
| | |
| +-------------------+ +-------------------+ +-------------------+ +-------------------+ +---------+---------+ |
| | Audio Transcription| | OCR Tools | | Text Classify | | CLI Tools | | Dynamic API | |
| | (Whisper, etc.) | | (Tesseract+LLM) | | | | (jq, rg, awk) | | (OpenAPI->Tool) | |
| +-------------------+ +-------------------+ +-------------------+ +-------------------+ +-------------------+ |
| |
| +-------------------+ +-------------------+ +-------------------+ +-------------------+ +-------------------+ |
| | Caching Service | | Analytics/Metrics | | Prompt Management | | Config Service | | Meta Tools/Refiner| |
| | (Memory/Disk/Redis| | (Cost/Usage Track)| | (Jinja2/Repo) | | (Loads .env) | | (list_tools etc.) | |
| +-------------------+ +-------------------+ +-------------------+ +-------------------+ +-------------------+ |
+------------------------------------------------------------------------------------------------------------------------+
委派请求流程（详细）
- Agent 决策：MCP Agent 判断需要某项特定能力（例如，总结长文本、提取 JSON、浏览 URL），并认为可以委派处理。
- 构造 MCP 请求：Agent 构造 MCP 工具调用请求，根据工具的架构指定
tool_name和所需的inputs（可能已通过list_tools发现该架构）。 - 向服务发送 HTTP POST 请求：Agent 通过 HTTP POST 将此请求（通常以 JSON 格式放在请求正文中）发送到 Ultimate MCP Server 的指定端点。
- 接收并解析请求：服务器的 Web 框架（FastAPI）接收请求。MCP Core 解析 JSON 请求正文，并根据通用 MCP 请求结构进行验证。
- 分发工具调用：MCP Core 在已注册工具的注册表中查找请求的
tool_name。 - 输入验证：服务器使用相应工具的输入架构（Pydantic 模型）验证请求中提供的
inputs。如果验证失败，则立即生成 MCP 错误响应。 - 工具执行上下文：可能会创建一个上下文对象，其中可能包含配置、对共享服务（如日志记录、缓存、分析）的访问权限等。
- 缓存检查：系统查询缓存服务。它根据
tool_name和经过验证的inputs生成缓存键。如果该键存在有效且未过期的缓存条目，则检索并返回缓存的响应（跳至第 14 步）。 - 执行工具逻辑：如果未命中缓存，则运行工具的主要执行逻辑：
- 大模型任务：如果该工具涉及调用大模型（例如，
completion、summarize_document、extract_json）：- 优化／路由逻辑会根据参数（
provider、model、provider_preference）和服务器配置选择提供商／模型。 - 提示词管理服务可能会使用模板格式化最终提示词。
- 提供商抽象层会为选定的提供商构造特定的 API 请求。
- 发起 API 调用，并处理可能的重试和超时。
- 接收并解析大模型响应。
- 优化／路由逻辑会根据参数（
- 专用工具任务：如果这是非大模型工具（例如，
read_file、browser_navigate、run_sql_query、run_ripgrep）：- 工具直接与相关系统交互（文件系统、Playwright 浏览器实例、数据库连接、子进程执行）。
- 执行安全检查（例如，允许访问的目录、SQL 清理占位符）。
- 获取操作结果。
- 大模型任务：如果该工具涉及调用大模型（例如，
- 成本计算：对于大模型任务，分析服务会根据输入／输出令牌数和提供商定价计算预估成本。对于其他任务，成本通常为零，除非它们消耗了特定的计量资源。
- 结果格式化：工具根据其定义的输出架构格式化结果（数据或错误消息）。
- 记录分析数据：分析服务会记录请求、响应（或错误）、执行时间、成本、使用的提供商／模型、缓存状态（命中／未命中）等。
- 更新缓存：如果操作成功，且此工具／请求启用了缓存，缓存服务会按计算出的 TTL 存储格式化后的响应。
- 构造 MCP 响应：MCP Core 将最终结果（来自缓存或执行结果）打包为标准 MCP 响应结构，其中包括
status、outputs、error（如有），以及可能包含的cost、usage_metadata。 - 向 Agent 返回 HTTP 响应：服务器将 MCP 响应作为 HTTP 响应发送回 Agent（通常状态为 200 OK，即使工具操作失败也是如此——MCP 请求本身已成功）。随后，Agent 解析此响应以确定工具调用结果。
🌍 实际应用场景
高级 AI Agent 能力
通过为 Claude 等 Agent 或自定义自主 Agent 提供以下工具，赋予它们执行复杂多模态任务的能力：
- 持久记忆与学习： 使用认知记忆系统，在长时间对话或任务中保持上下文。
- 网页交互与研究： 自动浏览网页、从网站提取数据、提交表单，并综合多个在线来源的信息。
- 数据分析与报告： 在 Excel 电子表格中创建、处理和分析数据；生成图表和报告。
- 数据库操作： 访问并查询企业数据库，根据 Agent 的目标检索或更新信息。
- 文档理解： 处理 PDF、图像（OCR），提取关键信息、总结长篇报告，并根据文档回答问题（RAG）。
- 知识图谱管理： 构建并查询有关特定领域、项目或实体的内部知识图谱。
- 多媒体处理： 转录会议录音或语音备忘录。
- 代码执行与分析： 使用命令行工具或专用代码工具（如果已添加）处理开发或数据任务。
- 外部服务集成： 与通过 OpenAPI 动态注册的其他公司 API 或公共 API 交互。
企业工作流自动化
构建利用 AI 推理和专用工具的复杂自动化流程：
- 智能文档处理流水线： 导入扫描件/PDF -> OCR -> 提取结构化数据（JSON）-> 验证数据 -> 分类文档类型 -> 路由到相应系统，或生成摘要供人工审核。
- 自动化研究助手： 给定主题 -> 搜索学术数据库（通过 Browser/API 工具）-> 下载相关论文（Browser/Filesystem）-> 对论文分块并生成摘要（文档工具）-> 提取关键发现（提取工具）-> 存入认知记忆 -> 生成综合报告。
- 财务报告自动化： 连接数据库（SQL 工具）-> 提取财务数据 -> 填充 Excel 模板（Excel 工具）-> 生成图表和差异分析 -> 发送报告邮件（如果已添加邮件工具）。
- 客户支持工单信息补充： 接收工单文本 -> 分类问题类型（分类工具）-> 搜索内部知识库和文档（RAG 工具）-> 起草建议回复 -> 通过 DB 或 API 工具从 CRM 获取客户详情并补充信息。
- 竞争对手监控： 安排浏览器自动化任务 -> 访问竞争对手网站/新闻源 -> 提取重要公告/价格变动 -> 总结发现 -> 提醒相关团队。
数据处理与集成
处理超越简单 ETL 的复杂数据任务：
- 非结构化数据转结构化数据： 从电子邮件、报告、聊天记录和产品评论中提取特定信息（JSON、表格）。
- 知识图谱创建： 处理文档语料库（例如，公司 Wiki、研究论文），构建实体关系图谱以查询洞察。
- 数据转换与清理： 使用 SQL 工具、Excel 自动化或本地文本处理工具（awk、sed），根据 LLM 指令执行复杂的数据处理。
- 自动数据分类： 对大型数据集应用文本分类工具（例如，对用户反馈分类、为新闻文章添加标签）。
- 语义数据搜索： 在内部文档上构建可搜索的向量索引，让用户或 Agent 能够按含义而非仅按关键词查找信息（RAG）。
研究与分析（科学、市场等）
使用 AI 驱动的工具为研究团队提供支持：
- 自动化文献搜索与综述： 使用浏览器/API 工具搜索数据库（PubMed、ArXiv 等），下载论文、拆分文本、生成摘要，并提取关键方法或结果。
- 比较分析： 使用多提供商补全或竞赛工具，比较不同模型如何根据研究数据解读或生成假设。
- 研究数据提取： 自动将已发表论文或报告中的结构化数据（参与者人数、p 值、结果）提取到数据库或电子表格中。
- 预算跟踪： 使用分析功能监控研究任务相关的大模型 API 费用。
- 持久化研究日志： 使用 Cognitive Memory 系统，在整个研究项目期间存储发现、假设、观察结果和推理步骤。
文档智能
构建全面的文档集理解系统：
- 端到端管道： OCR 扫描文档 -> 使用大模型增强文本 -> 提取预定义字段（提取工具）-> 对文档类型分类 -> 识别关键实体/关系 -> 生成摘要 -> 将文本和元数据编入可搜索系统（向量/SQL 数据库）。
金融分析与建模
为金融专业人士提供先进工具：
- AI 辅助建模： 使用自然语言指示 Excel 自动化工具创建复杂的金融模型、预测或估值分析。
- 数据集成： 通过浏览器自动化或 API 获取市场数据，并将其与数据库中的内部数据（SQL 工具）结合。
- 报告分析： 使用 RAG 或摘要工具快速理解篇幅较长的金融报告或申报文件。
- 情景测试： 以编程方式修改 Excel 模型中的输入，进行敏感性分析。
- 决策跟踪： 使用 Cognitive Memory 记录投资决策或分析背后的推理过程。
🔐 安全注意事项
部署和运行 Ultimate MCP Server 时，安全必须是首要考虑因素。请考虑以下方面：
-
🔑 API 密钥管理：
- 切勿将 API 密钥硬编码到源代码中，也不要将其提交到版本控制系统。
- 使用环境变量（本地开发时使用
.env文件、系统环境变量，或在生产环境中优先使用 HashiCorp Vault、AWS Secrets Manager、GCP Secret Manager 等密钥管理工具）。 - 确保
.env文件（如在本地使用）具有严格的文件权限（例如，chmod 600 .env），且只有运行服务的用户可以读取。 - 为开发和生产环境使用不同的密钥。
- 实施密钥轮换策略，并立即撤销疑似已泄露的密钥。
-
🌐 网络暴露与访问控制：
- 默认绑定到
127.0.0.1（SERVER_HOST），仅允许本地连接。只有在计划对外暴露服务时才更改为0.0.0.0，并且仅限在适当的网络控制措施保护下进行。 - 使用反向代理： 强烈建议在服务前部署反向代理（Nginx、Caddy、Traefik 等）。它可以处理 SSL/TLS 终止、实施访问控制（IP 允许列表、客户端证书身份验证、Basic Auth、OAuth2 代理集成），并提供隔离层。
- 防火墙规则： 配置主机或网络防火墙，将对
SERVER_PORT的访问限制为仅允许可信来源（例如，反向代理的 IP、特定应用服务器的 IP、VPN 网段）。
- 默认绑定到
-
👤 身份验证与授权：
- Ultimate MCP Server 本身可能没有内置的用户/Agent 身份验证。通常应在服务器之前的层（例如反向代理或 API 网关）处理身份验证。
- 确保只有经过授权的客户端（可信的 AI Agent、特定后端服务）可以向服务器端点发送请求。如有需要，可考虑使用由代理/网关管理的双向 TLS（mTLS）或 API 密钥/令牌。
- 如果工具提供不同级别的访问权限（例如，只读与读写文件系统），请考虑是否需要在服务器内部实现授权逻辑，或由外部管理。
-
🚦 速率限制与滥用防范：
- 在反向代理或 API 网关层，根据源 IP、API 密钥或其他标识符实施速率限制。这可以防止拒绝服务（DoS）攻击，并有助于控制过度 API 使用带来的成本（包括大模型和潜在的工具使用成本）。
- 监控使用模式，留意滥用迹象。
-
🛡️ 输入验证与清理：
- MCP 提供了结构化格式，但对于根据用户/Agent 输入与外部系统交互的工具，仍需格外注意：
- 文件系统工具： 至关重要的是，要严格配置
ALLOWED_DIRS。严格验证并规范化所有路径输入，防止目录遍历（../）。确保服务器进程以最小权限运行。 - SQL 工具： 正确使用参数化查询或 ORM（如 SQLAlchemy），防止 SQL 注入漏洞。避免直接根据 Agent 输入构造 SQL 字符串。
- 浏览器工具： 谨慎使用可执行任意 JavaScript 的工具（
browser_evaluate_script）。尽可能避免直接运行基于不可信 Agent 输入的脚本。Playwright 的沙盒机制有所帮助，但并非万无一失。 - CLI 工具： 清理传递给
run_ripgrep、run_jq等工具的参数，以防止命令注入，尤其是在构造复杂命令字符串时。使用安全的方式传递输入数据（例如 stdin）。
- 文件系统工具： 至关重要的是，要严格配置
- 使用 Pydantic 模式验证所有工具输入的数据类型和约束。
- MCP 提供了结构化格式，但对于根据用户/Agent 输入与外部系统交互的工具，仍需格外注意：
-
📦 依赖项安全：
- 定期使用
uv pip install --upgrade ...或uv sync更新依赖项，修补第三方库（FastAPI、Pydantic、Playwright、数据库驱动程序等）中的已知漏洞。 - 使用安全扫描工具（
pip-audit、GitHub Dependabot、Snyk）自动识别pyproject.toml或requirements.txt中存在漏洞的依赖项。
- 定期使用
-
📄 日志安全：
- 请注意，
DEBUG级别的日志可能会记录敏感信息，包括完整提示词、API 响应、文件内容或数据中包含的密钥。为生产环境正确配置LOG_LEVEL（通常使用INFO或WARNING更安全）。 - 确保日志文件（如果使用
LOG_TO_FILE）具有适当的权限，并考虑日志轮转和保留策略。避免记录原始 API 密钥。
- 请注意，
-
⚙️ 工具特定安全：
- 检查每个已启用工具的具体安全影响。它是否允许写入文件、执行代码或访问数据库？确保配置（如
ALLOWED_DIRS、权限受限的数据库凭据）遵循最小权限原则。禁用环境中不需要或无法充分保障安全的工具。
- 检查每个已启用工具的具体安全影响。它是否允许写入文件、执行代码或访问数据库？确保配置（如
📃 许可证
本项目采用 MIT License（含 OpenAI/Anthropic Rider）授权，详情请参阅 LICENSE 文件。
🙏 致谢
本项目基于许多出色的开源项目和服务构建。特别感谢：
- Model Context Protocol (MCP)，提供基础概念和协议规范。
- FastAPI 团队，提供高性能 Web 框架。
- Pydantic 开发者，提供可靠的数据验证和设置管理功能。
- Rich 库，提供美观且信息丰富的终端输出。
- Astral 的 uv，提供极速的 Python 包安装和解析功能。
- Microsoft 的 Playwright 团队，提供强大的浏览器自动化框架。
- OpenPyXL 维护者，提供 Excel 文件处理功能。
- SQLAlchemy 开发者，提供数据库工具包。
Tesseract、ripgrep、jq、awk、sed等集成工具的开发者。- 所有 LLM 提供商（OpenAI、Anthropic、Google、DeepSeek、xAI 等），让我们能通过 API 使用其强大的模型。
- 更广泛的 Python 和开源社区。
本 README 提供了全面的概述。如需了解特定工具参数、高级配置选项和详细的实现说明，请参阅项目中的源代码和各个工具文档。
运行服务
使用 CLI 启动服务：
# Start in default stdio mode
umcp run
# Start in streamable-http mode for web interfaces or remote clients (recommended)
umcp run --transport-mode shttp
# Or use the shortcut:
umcp run -t shttp
# Run on a specific host and port (streamable-http mode)
umcp run -t shttp --host 0.0.0.0 --port 8080
评论
支持 Markdown
一个开放协议，规定了 AI 应用如何以统一方式连接外部工具与数据源。客户端负责与模型对话，MCP 服务负责提供工具、资源和提示模板，两者通过标准消息通信，同一个服务可以被不同客户端复用。