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

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

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

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

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

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

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

/orchestrator-on

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

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

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

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

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

# 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

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

# 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

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

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

<project-root>/generated/unified_api.py

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