MCP Tools Orchestrator

Compose tools from multiple MCP servers into unified Python policies

MCP Tools Orchestrator is a meta-MCP server that enables "Code as Policies" across your entire MCP ecosystem. It automatically discovers tools from all your connected MCP servers and provides a unified Python API for writing complex, multi-server workflows.

🎯 What Problem Does This Solve?

Traditional MCP Usage:

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

With MCP Tools Orchestrator:

# 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!

Benefits:

• ✅ 10-100x faster: One execution instead of N round-trips
• ✅ Complex logic: Loops, conditionals, error handling in Python
• ✅ Multi-server workflows: Use tools from ANY server in one policy
• ✅ Immediate feedback: Scripts see results and adapt without agent involvement

---

🏗️ Architecture

Hybrid Design: No Duplicate Server Processes

MCP Tools Orchestrator leverages mcp-client's existing server connections via HTTP IPC instead of creating its own connections. This prevents duplicate server processes and resource conflicts.

┌─────────────────────────────────────────────────────────────┐
│                     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.)         │
└────────────────────────────────────────────┘

Key Points:

• Client manages all MCP server connections
• Orchestrator never connects directly to MCP servers
• Policy scripts call client's IPC server via HTTP
• Single process per MCP server (no duplicates!)

---

📦 Installation

Prerequisites

1. Python 3.10+ (developed with Python 3.13)
2. mcp-client with IPC support (see mcp-client-example)
3. uv package manager

Install MCP Tools Orchestrator

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

---

⚙️ Configuration

Step 1: Configure mcp-client

Add mcp-tools-orchestrator to your mcp_config.json (typically in ~/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"]
    }
  }
}

Note: The client will automatically:

• Start an IPC HTTP server on a random port
• Set MCP_CLIENT_IPC_URL environment variable for orchestrator
• Pass the IPC URL when spawning mcp-tools-orchestrator

Step 2: Configure Orchestrator's Server List

Create mcp_servers_config.json in the orchestrator directory:

{
  "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"]
    }
  }
}

Purpose: This config is used only for introspection (extracting tool signatures). Orchestrator doesn't spawn these servers - the client does!

---

🚀 Usage

1. Basic Workflow

Start the mcp-client with orchestrator enabled:

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

That's it! The unified API is automatically generated when mcp-tools-orchestrator starts. No manual generation step needed.

Enable orchestrator mode (optional but recommended):

/orchestrator-on

This hides all direct tools and shows only orchestrator tools, reducing context pollution.

2. Ask the Agent to Write a Policy

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

The agent will use execute_composed_code with a Python script:

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. Available Orchestrator Tools

MCP Tools Orchestrator provides 4 tools to the agent:

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

Execute Python code with access to ALL tools from ALL connected servers.

Returns: {output: str, returncode: int, status: str}

list_available_tools()

Get a structured view of all available tools with their signatures.

Returns: {servers: {...}, total_servers: int, total_tools: int}

refresh_tools()

Re-discover tools from all servers (useful if servers were updated).

Returns: {status: str, server_count: int, tool_count: int}

get_api_documentation()

Get documentation about the generated unified API.

Returns: str (formatted documentation)

---

📚 How It Works

1. Initialization (When Orchestrator Starts)

# 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 Generation via Introspection

# 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)

Why introspection?

• Previous approach used JSON schemas → functions had no parameters
• Introspection uses Python's inspect.signature() → accurate signatures
• Each server introspected in its own environment → no dependency conflicts

3. Generated API Structure

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

_IPC_URL = "http://localhost:<random_port>"  # 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. Code Execution Flow

# 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"
        }

---

📁 Project Structure

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

Active Files (Clean Architecture):

• server.py - Main MCP server
• src/mcp_tools_orchestrator/api_generator.py - API generation via introspection
• src/mcp_tools_orchestrator/introspect_server.py - Isolated introspection script
• src/mcp_tools_orchestrator/code_executor.py - Policy code execution

Generated Files:

• generated/unified_api.py - Auto-generated on every server startup (no manual steps needed)

---

🎓 Example Policies

Simple Grasp with Verification

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")

Multi-Server Workflow with Error Recovery

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")

Complex Assembly with Resource Tracking

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

More examples in the examples/ directory!

---

🔧 Development

Running in Development

# 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:<port>"
python server.py

Note: The API is automatically generated on startup. The sections below are for development/debugging only.

Regenerating the API Manually (Development Only)

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

Testing Introspection

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

---

🚨 Important Notes

Environment Variables

Required:

• MCP_CLIENT_IPC_URL - Set automatically by mcp-client when spawning orchestrator

Optional:

• MCP_CLIENT_OUTPUT_DIR - Shared outputs directory (set by client)

Introspection Requirements

Each server in mcp_servers_config.json must:

1. Be a valid Python script
2. Use MCP decorators (@mcp.tool())
3. Have type-hinted function signatures
4. Be runnable in its specified Python environment

Python Version Compatibility

Developed with: Python 3.13 Minimum required: Python 3.10

The .python-version file specifies 3.13 for consistency. If you encounter issues, ensure your environment matches or update .python-version to your Python version.

Generated API Location

The unified API is always generated at:

<project-root>/generated/unified_api.py

This path is determined by server.py:

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

---

💡 Benefits Over Alternatives

vs. Manual Tool Calls (Traditional MCP)

Aspect	Manual Tool Calls	MCP Tools Orchestrator
Speed	~2s per tool call	All tools in one execution
Complexity	Limited to agent's planning	Full Python: loops, conditionals, functions
Knowledge	Agent must track state	Script has full context
Latency	N round-trips	1 execution

vs. Per-Server Custom APIs

Aspect	Custom APIs	MCP Tools Orchestrator
Maintenance	Write API for each server	Auto-generated
Updates	Manual sync	Auto-refresh
Cross-server	Complex coordination	Natural in policy code
Type safety	Manual typing	Auto-extracted from servers

---

🐛 Known Limitations

1. Abort Signal Handling

   • Client-side abort functionality is fully implemented (press 'a' to abort)
   • Orchestrator's generated API needs update to detect [ABORTED] prefix
   • Scripts currently treat abort as normal error instead of immediate termination

2. Introspection Edge Cases

   • Bash-wrapped commands require parsing (works but fragile)
   • Very large servers may timeout during introspection

3. Error Context

   • Stack traces from policy scripts can be verbose
   • Errors don't always indicate which server/tool failed

---

🗺️ Future Enhancements

• [ ] Implement proper abort signal detection in unified_api.py
• [ ] Cache introspection results for faster startup
• [ ] WebSocket support for lower IPC latency
• [ ] Script library/registry for reusable policies
• [ ] Better error messages with server/tool context
• [ ] Support for streaming tool results
• [ ] Interactive debugging mode

---

📄 License

MIT License - See LICENSE file for details

---

👤 Author

Aldrin Inbaraj Email: aaugus11@asu.edu GitHub: [Your GitHub Profile]

---

🙏 Acknowledgments

• Built on the Model Context Protocol (MCP)
• Uses FastMCP for server implementation
• Inspired by "Code as Policies" paradigm from robotics research

---

📞 Support

For issues, questions, or contributions:

1. Review the documentation in this README

2. Check example policies in examples/

3. Open an issue on GitHub with:

   • Clear description of the problem
   • Relevant logs/error messages
   • Steps to reproduce

---

Happy Policy Writing! 🚀