MCP工具编排器

将来自多个MCP服务器的工具组合到统一的Python策略中

MCP Tools Orchestrator是一个元MCP服务器，可在整个MCP生态系统中启用“代码即策略”。它自动从所有连接的MCP服务器中发现工具，并提供统一的Python API，用于编写复杂的多服务器工作流。

🎯 这解决了什么问题？

传统MCP用法：

Agent: I'll call tool A
→ Wait for result
Agent: Based on A, I'll call tool B
→ Wait for result
Agent: Based on B, I'll call tool C
→ Wait for result

使用MCP工具编排器：

# Agent writes one policy script that orchestrates everything
for attempt in range(10):
    result_a = server1__tool_a()
    if result_a["success"]:
        result_b = server2__tool_b(result_a["data"])
        if result_b["status"] == "ready":
            server3__tool_c()
            break
    # Complex logic with loops, conditionals, error handling!

优点：

✅ 快10-100倍：一次执行，而不是N次往返
✅ 复杂逻辑：Python中的循环、条件、错误处理
✅ 多服务器工作流：在一个策略中使用来自任何服务器的工具
✅ 即时反馈：脚本在没有代理参与的情况下看到结果并进行调整

______________________________________________________________________

🏗️ 建筑

混合设计：没有重复的服务器进程

MCP工具编排器利用 mcp客户端的现有服务器连接 通过HTTP IPC，而不是创建自己的连接。这可以防止重复的服务器进程和资源冲突。

┌─────────────────────────────────────────────────────────────┐
│                     mcp-client (CLI)                         │
│  ┌──────────────────────────────────────────────────────┐   │
│  │  Agent (Claude/GPT)                                   │   │
│  │  Calls: mcp-tools-orchestrator__execute_composed_code(script)  │   │
│  └──────────────────────────────────────────────────────┘   │
│         ↓                                                    │
│  ┌──────────────────────────────────────────────────────┐   │
│  │  IPC Server (HTTP)                                    │   │
│  │  http://localhost:random_port                         │   │
│  │  Routes tool calls to appropriate MCP servers         │   │
│  └──────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────┘
         ↓ (MCP stdio)              ↑ (HTTP IPC)
┌──────────────────────┐    ┌────────────────┐
│   mcp-tools-orchestrator │    │ Python Script  │
│   (server.py)        │    │ (user policy)  │
│                      │    │                │
│ 1. Generates API     │    │ from unified_  │
│ 2. Executes scripts  │←───│ api import *   │
└──────────────────────┘    └────────────────┘

                ↓ (HTTP POST /call_tool)

┌────────────────────────────────────────────┐
│        Actual MCP Servers                  │
│  (ros-mcp-server, isaac-sim, etc.)         │
└────────────────────────────────────────────┘

要点：

客户端管理全部 MCP服务器连接
编排器从不直接连接到MCP服务器
策略脚本通过HTTP调用客户端的IPC服务器
每个MCP服务器只有一个进程（没有重复！）

______________________________________________________________________

📦 安装

先决条件

Python 3.10+ （使用Python 3.13开发）
mcp客户端 支持IPC（请参见 mcp客户端示例)
紫外线 包管理器

安装MCP工具编排器

cd /path/to/mcp-tools-orchestrator
uv sync

______________________________________________________________________

⚙️ 配置

步骤1：配置mcp客户端

将mcp工具编排器添加到您的 mcp_config.json （通常在 ~/Documents/mcp-client-example/):

{
  "mcpServers": {
    "mcp-tools-orchestrator": {
      "disabled": false,
      "timeout": 60,
      "type": "stdio",
      "command": "/path/to/mcp-tools-orchestrator/.venv/bin/python",
      "args": ["/path/to/mcp-tools-orchestrator/server.py"]
    },
    "ros-mcp-server": {
      "disabled": false,
      "command": "bash",
      "args": ["-c", "source /opt/ros/humble/setup.bash && python server.py"]
    },
    "isaac-sim": {
      "disabled": false,
      "command": "python",
      "args": ["/path/to/isaac-sim-mcp/server.py"]
    }
  }
}

备注：客户端将自动：

在随机端口上启动IPC HTTP服务器
集 MCP_CLIENT_IPC_URL 编排器的环境变量
生成mcp工具编排器时传递IPC URL

步骤2：配置编排器的服务器列表

创建 mcp_servers_config.json 在编排器目录中：

{
  "mcpServers": {
    "ros-mcp-server": {
      "command": "bash",
      "args": [
        "-c",
        "source /opt/ros/humble/setup.bash && /home/user/.pyenv/versions/3.10.12/bin/python /path/to/server.py"
      ]
    },
    "isaac-sim": {
      "command": "/home/user/.pyenv/versions/3.10.12/bin/python",
      "args": ["/path/to/isaac-sim-mcp/server.py"]
    },
    "Resources": {
      "command": "/home/user/.pyenv/versions/3.10.12/bin/python",
      "args": ["/path/to/grasp_assembly_server/server.py"]
    }
  }
}

目的：使用此配置 只是为了内省 （提取工具签名）。Orchestrator不会生成这些服务器，而是由客户端生成！

______________________________________________________________________

🚀 用法

1.基本工作流程

在启用编排器的情况下启动mcp客户端：

cd ~/path/to/mcp-client
mcp-client --all  # Connects to all enabled servers including orchestrator

就是这样！ 统一的API是 自动生成 当mcp工具编排器启动时。无需手动生成步骤。

启用编排器模式（可选但推荐）：

/orchestrator-on

这隐藏了所有直接工具，只显示编排工具，减少了上下文污染。

2.要求代理人写一份保单

User: Write a script to try grasping 5 different objects and report the success rate

代理人将使用 execute_composed_code 使用Python脚本：

from unified_api import *

success_count = 0
total = 5

for i in range(total):
    # Move to grasp position
    result = ros_mcp_server__move_to_grasp(
        object_name=f"object_{i}",
        grasp_id=0,
        mode="sim",
        move_to_object=True
    )

    if result.get("success"):
        # Close gripper
        ros_mcp_server__control_gripper("close", mode="sim")

        # Verify grasp
        verify = ros_mcp_server__verify_grasp(f"object_{i}", mode="sim")
        if verify.get("result") == "SUCCESS":
            success_count += 1
            print(f"✓ Object {i} grasped successfully")
        else:
            print(f"✗ Object {i} grasp failed")

print(f"\nSuccess rate: {success_count}/{total} ({success_count/total*100:.1f}%)")

3.可用的编排器工具

MCP工具编排器为代理提供4个工具：

`execute_composed_code(code: str, timeout: int = 3600)`

通过从所有连接的服务器访问所有工具来执行Python代码。

退货: {output: str, returncode: int, status: str}

`list_available_tools()`

获取所有可用工具及其签名的结构化视图。

退货: {servers: {...}, total_servers: int, total_tools: int}

`refresh_tools()`

从所有服务器中重新发现工具（如果服务器已更新，则很有用）。

退货: {status: str, server_count: int, tool_count: int}

`get_api_documentation()`

获取有关生成的统一API的文档。

退货: str （格式化文档）

______________________________________________________________________

📚 运作原理

1.初始化（当编排器启动时）

# In server.py
async def initialize():
    # 1. Check for client IPC URL
    client_ipc_url = os.getenv("MCP_CLIENT_IPC_URL")  # Set by client

    # 2. Generate unified API using introspection
    generator = UnifiedAPIGenerator()
    generator.generate_api_from_config(
        "mcp_servers_config.json",
        "generated/unified_api.py",
        client_ipc_url
    )

    # 3. Initialize code executor
    executor = CodeExecutor("generated/unified_api.py", client_ipc_url)

2.通过反思生成API

# In api_generator.py
class UnifiedAPIGenerator:
    def generate_api_from_config(self, config_path, output_path, ipc_url):
        # For each server in config:
        for server_name, server_config in config["mcpServers"].items():

            # 1. Extract Python path and server script path
            python_path, server_path = self._extract_paths(server_config)

            # 2. Run introspection in isolated subprocess
            #    (avoids dependency conflicts between servers)
            tools = subprocess.run([
                python_path,
                "introspect_server.py",  # Isolated introspection script
                server_path,
                server_name
            ])

            # 3. Parse tool signatures (params, types, defaults, docstrings)
            all_tools[server_name] = parse_tools(tools.stdout)

        # 4. Generate unified_api.py using Jinja2 template
        self._generate_api_file(all_tools, output_path, ipc_url)

为什么要内省？

以前的方法使用JSON模式→ 功能有 没有参数
内省使用Python inspect.signature() → 准确的签名
每个服务器都在自己的环境中进行了反思→ 无依赖冲突

3.生成的API结构

# In generated/unified_api.py (auto-generated)
import requests

_IPC_URL = "http://localhost:"  # Client's IPC server (set dynamically)

# Tools from ros-mcp-server
def ros_mcp_server__move_to_grasp(
    object_name: str,
    grasp_id: int,
    mode: str = "sim",
    move_to_object: bool = False,
    move_to_safe_height: bool = False
) -> dict:
    """Move to grasp position..."""
    return _call_tool("ros-mcp-server", "move_to_grasp", {
        "object_name": object_name,
        "grasp_id": grasp_id,
        "mode": mode,
        "move_to_object": move_to_object,
        "move_to_safe_height": move_to_safe_height
    })

# Helper function
def _call_tool(server: str, tool: str, arguments: dict) -> dict:
    response = requests.post(
        f"{_IPC_URL}/call_tool",
        json={"server": server, "tool": tool, "arguments": arguments},
        timeout=300
    )
    return response.json()

4.代码执行流程

# In code_executor.py
class CodeExecutor:
    def execute_code(self, user_code: str, timeout: int) -> dict:
        # 1. Wrap user code with imports
        wrapped = f"""
import sys
sys.path.insert(0, '{self.api_dir}')
from unified_api import *

{user_code}
"""

        # 2. Create temp file and execute in subprocess
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py') as f:
            f.write(wrapped)
            result = subprocess.run(
                [self.venv_python, f.name],
                capture_output=True,
                timeout=timeout,
                env={"MCP_ORCHESTRATOR_IPC_URL": self.client_ipc_url}
            )

        # 3. Return output and status
        return {
            "output": result.stdout,
            "error": result.stderr,
            "returncode": result.returncode,
            "status": "success" if result.returncode == 0 else "error"
        }

______________________________________________________________________

📁 项目结构

mcp-tools-orchestrator/
├── server.py                         # Main FastMCP server entry point
│
├── src/mcp_tools_orchestrator/
│   ├── api_generator.py              # Introspection-based API generator
│   ├── introspect_server.py          # Isolated server introspection script
│   ├── code_executor.py              # Executes policy code in subprocess
│   ├── __init__.py                   # Package initialization
│   └── py.typed                      # Type hints marker (PEP 561)
│
├── generated/
│   └── unified_api.py                # Auto-generated API (63 tools from 3 servers)
│
├── examples/
│   ├── simple_grasp.py               # Basic grasping workflow
│   ├── multi_server_workflow.py      # Cross-server orchestration
│   └── error_recovery.py             # Error handling patterns
│
├── mcp_servers_config.json           # Server config for introspection
├── pyproject.toml                    # Project metadata and dependencies
├── uv.lock                           # Locked dependencies
│
├── README.md                         # This file
│
├── .python-version                   # Python 3.13 (for pyenv)
└── .gitignore                        # Git ignore rules

活动文件（干净架构）：

server.py -主MCP服务器
src/mcp_tools_orchestrator/api_generator.py -通过内省生成API
src/mcp_tools_orchestrator/introspect_server.py -独立内省脚本
src/mcp_tools_orchestrator/code_executor.py -策略代码执行

生成的文件：

generated/unified_api.py - 每次服务器启动时自动生成 （无需手动步骤）

______________________________________________________________________

🎓 政策示例

简单掌握与验证

from unified_api import *

# Move to home position
ros_mcp_server__move_home()

# Open gripper
ros_mcp_server__control_gripper("open", mode="sim")

# Move to grasp
ros_mcp_server__move_to_grasp(
    object_name="block_1",
    grasp_id=0,
    mode="sim",
    move_to_object=True
)

# Close gripper
ros_mcp_server__control_gripper("close", mode="sim")

# Move to safe height
ros_mcp_server__move_to_grasp(
    object_name="block_1",
    grasp_id=0,
    mode="sim",
    move_to_safe_height=True
)

# Verify grasp
result = ros_mcp_server__verify_grasp("block_1", mode="sim")
if result["result"] == "SUCCESS":
    print("✓ Grasp successful!")
else:
    print("✗ Grasp failed")

具有错误恢复功能的多服务器工作流

from unified_api import *

# Save scene state before attempting grasps
scene_id = isaac_sim__save_scene_state()
print(f"Saved scene state: {scene_id}")

# Try multiple grasp poses
for grasp_id in range(5):
    print(f"\nAttempting grasp {grasp_id}...")

    # Move to grasp
    ros_mcp_server__move_to_grasp(
        object_name="gear",
        grasp_id=grasp_id,
        mode="sim",
        move_to_object=True
    )

    # Close gripper
    ros_mcp_server__control_gripper("close", mode="sim")

    # Move to safe height
    ros_mcp_server__move_to_grasp(
        object_name="gear",
        grasp_id=grasp_id,
        mode="sim",
        move_to_safe_height=True
    )

    # Verify
    result = ros_mcp_server__verify_grasp("gear", mode="sim")

    if result["result"] == "SUCCESS":
        print(f"✓ Grasp {grasp_id} succeeded!")
        break
    else:
        print(f"✗ Grasp {grasp_id} failed, restoring scene...")
        isaac_sim__restore_scene_state()
else:
    print("All grasp attempts failed")

具有资源跟踪功能的复杂程序集

from unified_api import *

# Get successful grasp configurations from resource server
assembly_id = "3"
configs = Resources__get_object_grasp_configs_by_result(
    assembly_id=assembly_id,
    object_name="gear",
    result="SUCCESS"
)

print(f"Found {len(configs)} successful grasp configs")

# Try each successful configuration
for config in configs:
    grasp_id = config["grasp_id"]
    gripper_state = config["gripper_state"]

    print(f"\nTrying grasp {grasp_id} with gripper {gripper_state}")

    # Set gripper state BEFORE grasping (important!)
    ros_mcp_server__control_gripper(gripper_state, mode="sim")

    # Attempt grasp
    ros_mcp_server__move_to_grasp(
        object_name="gear",
        grasp_id=grasp_id,
        mode="sim",
        move_to_object=True
    )

    # Verify
    result = ros_mcp_server__verify_grasp("gear", mode="sim")

    if result["result"] == "SUCCESS":
        print(f"✓ Successfully grasped using config {grasp_id}")

        # Save this trial to resource server
        Resources__write_assembly_resource(
            assembly_id=assembly_id,
            object_name="gear",
            sequence_id=1,
            assembled_into="base",
            tools_trials=[{
                "trial_id": 1,
                "grasp_id": grasp_id,
                "gripper_state": gripper_state,
                "tools": ["move_to_grasp", "verify_grasp"],
                "result": "SUCCESS"
            }]
        )
        break

更多示例请参见 examples/ 目录！

______________________________________________________________________

🔧 发展

在发展中奔跑

# The server requires MCP_CLIENT_IPC_URL to be set
# Normally set by mcp-client, but for testing:
export MCP_CLIENT_IPC_URL="http://localhost:
"
python server.py

注： API是在启动时自动生成的。以下部分仅用于开发/调试。

手动重新生成API（仅限开发）

python src/mcp_tools_orchestrator/api_generator.py \
    mcp_servers_config.json \
    generated/unified_api.py \
    http://localhost:

测试反思

# Test introspection of a specific server
python src/mcp_tools_orchestrator/introspect_server.py \
    /path/to/server.py \
    server-name

______________________________________________________________________

🚨 重要提示

环境变量

必修的：

MCP_CLIENT_IPC_URL -生成编排器时由mcp客户端自动设置

可选：

MCP_CLIENT_OUTPUT_DIR -共享输出目录（由客户端设置）

反思要求

每台服务器 mcp_servers_config.json 必须：

成为有效的Python脚本
使用MCP装饰器(@mcp.tool())
具有类型暗示的函数签名
可在指定的Python环境中运行

Python版本兼容性

与开发：Python 3.13 最低要求：Python 3.10

这 .python-version 文件为一致性指定了3.13。如果遇到问题，请确保您的环境匹配或更新 .python-version 你的Python版本。

生成的API位置

统一的API始终在以下位置生成：


/generated/unified_api.py

此路径由以下因素决定 server.py:

script_dir = Path(__file__).parent  # Repository root
generated_dir = script_dir / "generated"

______________________________________________________________________

💡 替代方案的优势

与手动工具调用（传统MCP）相比

特性	手动工具调用	MCP工具编排器
速度	每次工具调用约2秒	一次执行所有工具
复杂性	仅限于代理的规划	完整的Python：循环、条件、函数
知识	代理必须跟踪状态	脚本具有完整上下文
延迟	N次往返	1次执行

与每服务器自定义API相比

特性	自定义API	MCP工具编排器
维护	为每个服务器编写API	自动生成
更新	手动同步	自动刷新
跨服务器	复杂的协调	在政策代码中很自然
类型安全性	手动键入	从服务器自动提取

______________________________________________________________________

🐛 已知限制

中止信号处理

- 客户端中止功能已完全实现（按“a”中止） - Orchestrator生成的API需要更新才能检测 [ABORTED] 前缀 - 脚本当前将中止视为正常错误，而不是立即终止

反思边缘案例

- Bash包装的命令需要解析（可以工作，但很脆弱） - 超大型服务器在自检过程中可能会超时

错误上下文

- 策略脚本中的堆栈跟踪可能很冗长 - 错误并不总是表明哪个服务器/工具失败

______________________________________________________________________

🗺️ 未来的增强功能

\[\]在中执行适当的中止信号检测 unified_api.py
\[\]缓存自检结果以加快启动速度
\[\]WebSocket支持降低IPC延迟
\[\]可重用策略的脚本库/注册表
\[\]更好的服务器/工具上下文错误消息
\[\]支持流媒体工具结果
\[\]交互式调试模式

______________________________________________________________________

📄 许可证

MIT许可证-有关详细信息，请参阅许可证文件

______________________________________________________________________

👤 作者

奥尔德林·因巴拉杰 电子邮件：aaugus11@asu.edu GitHub：\[您的GitHub个人资料\]

______________________________________________________________________

🙏 致谢

建立在模型上下文协议（MCP）
用途 FastMCP 用于服务器实现
受机器人研究“代码即政策”范式的启发

______________________________________________________________________

📞 支持

对于问题、疑问或贡献：

查看此README中的文档

查看中的示例策略 examples/

在GitHub上打开一个问题：

- 问题的清晰描述 - 相关日志/错误消息 - 重现步骤

______________________________________________________________________

快乐的政策写作！ 🚀

MCP Tools Orchestrator

快速接入

详细介绍

MCP工具编排器

🎯 这解决了什么问题？

🏗️ 建筑

混合设计：没有重复的服务器进程

📦 安装

先决条件

安装MCP工具编排器

⚙️ 配置

步骤1：配置mcp客户端

步骤2：配置编排器的服务器列表

🚀 用法

1.基本工作流程

2.要求代理人写一份保单

3.可用的编排器工具

execute_composed_code(code: str, timeout: int = 3600)

list_available_tools()

refresh_tools()

get_api_documentation()

📚 运作原理

1.初始化（当编排器启动时）

2.通过反思生成API

3.生成的API结构

4.代码执行流程

📁 项目结构

🎓 政策示例

简单掌握与验证

具有错误恢复功能的多服务器工作流

具有资源跟踪功能的复杂程序集

🔧 发展

在发展中奔跑

手动重新生成API（仅限开发）

测试反思

🚨 重要提示

环境变量

反思要求

Python版本兼容性

生成的API位置

💡 替代方案的优势

与手动工具调用（传统MCP）相比

与每服务器自定义API相比

🐛 已知限制

🗺️ 未来的增强功能

📄 许可证

👤 作者

🙏 致谢

📞 支持

目录标签

接入字段

权限和风险

来源信息

继续浏览同类 MCP

`execute_composed_code(code: str, timeout: int = 3600)`

`list_available_tools()`

`refresh_tools()`

`get_api_documentation()`