class RoutedAgent:
def __init__(self, server: MCPToolServer, router: HybridMCPRouter, model: str):
self.server = server
self.router = router
self.model = model
def discover_exposed_tools(self, exposed_tool_names: List[str]) -> List[Dict[str, Any]]:
return [t for t in self.server.tools_list() if t["name"] in exposed_tool_names]
def plan(self, task: str, exposed_tools: List[Dict[str, Any]]) -> PlanOutput:
instructions = """
You are a planning agent in an MCP-like architecture.
You can only use the exposed tools.
Decide whether tools are needed.
Return strict JSON only with keys:
requires_tools: boolean
tool_calls: array of objects with tool_name and arguments
direct_answer_allowed: boolean
planner_note: string
Rules:
- Use at most 3 tool calls.
- Only call tools from the exposed list.
- Arguments must match each tool's input schema conceptually.
- Prefer calling vector_retrieve for conceptual local knowledge.
- Prefer calling web_search for recent or external information.
- Prefer dataset_loader if the user asks about a named built-in dataset.
- Prefer python_exec only when computation or code execution is genuinely useful.
- Do not fabricate unavailable tools.
"""
prompt = f"""
USER TASK:
{task}
EXPOSED TOOLS:
{json.dumps(exposed_tools, indent=2)}
Return JSON only.
"""
obj = llm_json(instructions, prompt)
raw_tool_calls = obj.get("tool_calls", [])
parsed_calls = []
allowed = {t["name"] for t in exposed_tools}
for call in raw_tool_calls[:MAX_TOOL_CALLS]:
name = call.get("tool_name", "")
args = call.get("arguments", {})
if name in allowed and isinstance(args, dict):
parsed_calls.append(ToolCall(tool_name=name, arguments=args))
return PlanOutput(
requires_tools=bool(obj.get("requires_tools", False) or parsed_calls),
tool_calls=parsed_calls,
direct_answer_allowed=bool(obj.get("direct_answer_allowed", False)),
planner_note=obj.get("planner_note", ""),
)
def run_tools(self, tool_calls: List[ToolCall]) -> List[ToolResult]:
results = []
for tc in tool_calls:
result = self.server.tools_call(tc.tool_name, tc.arguments)
results.append(result)
return results
def answer(self, task: str, route: RouteDecision, exposed_tools: List[Dict[str, Any]], plan: PlanOutput, results: List[ToolResult]) -> str:
instructions = """
You are the final answering agent in an MCP-style routed tool system.
Use the routed tools and returned tool outputs to answer the user.
Be concrete, concise, and technically correct.
If tool outputs are partial, say so.
Do not mention hidden tools that were not exposed.
"""
tool_result_payload = [r.model_dump() for r in results]
prompt = f"""
USER TASK:
{task}
ROUTE DECISION:
{route.model_dump_json(indent=2)}
EXPOSED TOOLS:
{json.dumps(exposed_tools, indent=2)}
PLAN:
{plan.model_dump_json(indent=2)}
TOOL RESULTS:
{json.dumps(tool_result_payload, indent=2)}
Now answer the user clearly.
"""
resp = client.responses.create(
model=self.model,
input=prompt,
instructions=instructions,
temperature=0.2
)
return resp.output_text
def run(self, task: str, verbose: bool = True) -> Dict[str, Any]:
route = self.router.route(task)
exposed_tools = self.discover_exposed_tools(route.selected_tools)
plan = self.plan(task, exposed_tools)
results = self.run_tools(plan.tool_calls) if plan.requires_tools else []
final_answer = self.answer(task, route, exposed_tools, plan, results)
payload = {
"task": task,
"route_decision": route.model_dump(),
"exposed_tools": exposed_tools,
"plan": plan.model_dump(),
"tool_results": [r.model_dump() for r in results],
"final_answer": final_answer,
}
if verbose:
console.print(Panel.fit(f"USER TASK\n{task}", title="Input"))
pretty_tools_table(exposed_tools, "Tools Exposed By MCP Router")
console.print(Panel(route.rationale or "No rationale provided", title="Router Rationale"))
if route.policy_notes:
console.print(Panel("\n".join(f"- {x}" for x in route.policy_notes), title="Policy Notes"))
console.print(Panel(plan.planner_note or "No planner note provided", title="Planner Note"))
if results:
for r in results:
console.print(Panel.fit(RichJSON.from_data(r.model_dump()), title=f"Tool Result: {r.tool_name}"))
console.print(Panel(final_answer, title="Final Answer"))
return payload
def mcp_jsonrpc_tools_list(server: MCPToolServer) -> Dict[str, Any]:
return {
"jsonrpc": "2.0",
"id": 1,
"result": {
"tools": server.tools_list()
}
}
def mcp_jsonrpc_tools_call(server: MCPToolServer, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
result = server.tools_call(tool_name, arguments)
return {
"jsonrpc": "2.0",
"id": 2,
"result": result.model_dump()
}
router = HybridMCPRouter(server=server, model=MODEL)
agent = RoutedAgent(server=server, router=router, model=MODEL)
console.print(Panel.fit("MCP-STYLE TOOL DISCOVERY", title="Step 1"))
console.print(RichJSON.from_data(mcp_jsonrpc_tools_list(server)))
demo_tasks = [
"Explain how an MCP tool router should expose tools for an agent task about dynamic capability exposure.",
"Search the web for recent examples of MCP-related developments and summarize them.",
"Load the iris dataset, inspect its columns and basic stats, and tell me what kind of ML problem it is.",
"Retrieve local knowledge about context injection and router policies, then explain why restricting tool access helps agent performance.",
"Use Python to compute the average of [3, 5, 9, 10, 13] and then explain whether python execution was truly necessary.",
]
all_runs = []
for idx, task in enumerate(demo_tasks, start=1):
console.print(Panel.fit(f"DEMO RUN {idx}", title="=" * 10))
out = agent.run(task, verbose=True)
all_runs.append(out)
custom_task = "Design a routed MCP workflow for an AI research assistant that should use retrieval for local protocol knowledge and web search only when the task explicitly asks for recent information."
custom_run = agent.run(custom_task, verbose=True)
print("\nPROGRAMMATIC EXAMPLE: tools/list")
print(json.dumps(mcp_jsonrpc_tools_list(server), indent=2))
print("\nPROGRAMMATIC EXAMPLE: tools/call for vector_retrieve")
print(json.dumps(mcp_jsonrpc_tools_call(server, "vector_retrieve", {"query": "dynamic capability exposure in MCP routers", "top_k": 2}), indent=2))
print("\nPROGRAMMATIC EXAMPLE: tools/call for dataset_loader")
print(json.dumps(mcp_jsonrpc_tools_call(server, "dataset_loader", {"name": "iris", "n_rows": 5}), indent=2))
print("\nPROGRAMMATIC EXAMPLE: custom final answer")
print(custom_run["final_answer"])Credit: Source link
