MCP Tools Orchestrator
Integrates with ROS (Robot Operating System) via ros-mcp-server to enable robotic manipulation tasks such as moving to grasp positions, controlling grippers, and verifying grasps.
Click on "Install Server".
Wait a few minutes for the server to deploy. Once ready, it will show a "Started" state.
In the chat, type
@followed by the MCP server name and your instructions, e.g., "@MCP Tools OrchestratorExecute a policy that loops over all tools and retries failures."
That's it! The server will respond to your query, and you can continue using it as needed.
Here is a step-by-step guide with screenshots.
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 resultWith 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
Python 3.10+ (developed with Python 3.13)
mcp-client with IPC support (see mcp-client-example)
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_URLenvironment variable for orchestratorPass 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 orchestratorThat'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-onThis 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 rateThe 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 signaturesEach 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 rulesActive Files (Clean Architecture):
server.py- Main MCP serversrc/mcp_tools_orchestrator/api_generator.py- API generation via introspectionsrc/mcp_tools_orchestrator/introspect_server.py- Isolated introspection scriptsrc/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"
}]
)
breakMore 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.pyNote: 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:
Be a valid Python script
Use MCP decorators (
@mcp.tool())Have type-hinted function signatures
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.pyThis 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
Abort Signal Handling
Client-side abort functionality is fully implemented (press 'a' to abort)
Orchestrator's generated API needs update to detect
[ABORTED]prefixScripts currently treat abort as normal error instead of immediate termination
Introspection Edge Cases
Bash-wrapped commands require parsing (works but fragile)
Very large servers may timeout during introspection
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.pyCache 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:
Review the documentation in this README
Check example policies in
examples/Open an issue on GitHub with:
Clear description of the problem
Relevant logs/error messages
Steps to reproduce
Happy Policy Writing! π
This server cannot be installed
Maintenance
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
- Your AI Chatbot Just Exposed Your CEO's Salary to an InternBy Om-Shree-0709 on .Agent IdentityMCP SecurityOAuth Delegation
- Why MCP Servers Need Execution Sandboxing (And Why Your Current Stack Isn't Enough)By Om-Shree-0709 on .Agentic AiPrompt InjectionWebAssembly
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