Skip to main content
Glama
inbarajaldrin

MCP Tools Orchestrator

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


Related MCP server: MetaMCP MCP Server

πŸ—οΈ 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


πŸ“ž 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! πŸš€

F
license - not found
-
quality - not tested
C
maintenance

Maintenance

–Maintainers
–Response time
–Release cycle
–Releases (12mo)
Commit activity

Resources

Unclaimed servers have limited discoverability.

Looking for Admin?

If you are the server author, to access and configure the admin panel.

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/inbarajaldrin/mcp-tools-orchestrator'

If you have feedback or need assistance with the MCP directory API, please join our Discord server