class RoutedAgent:
def __init__(self, server: MCPToolServer, router: HybridMCPRouter, mannequin: str):
self.server = server
self.router = router
self.mannequin = mannequin
def discover_exposed_tools(self, exposed_tool_names: Listing[str]) -> Listing[Dict[str, Any]]:
return [t for t in self.server.tools_list() if t["name"] in exposed_tool_names]
def plan(self, activity: str, exposed_tools: Listing[Dict[str, Any]]) -> PlanOutput:
directions = """
You're a planning agent in an MCP-like structure.
You possibly can solely use the uncovered instruments.
Determine whether or not instruments are wanted.
Return strict JSON solely with keys:
requires_tools: boolean
tool_calls: array of objects with tool_name and arguments
direct_answer_allowed: boolean
planner_note: string
Guidelines:
- Use at most 3 device calls.
- Solely name instruments from the uncovered checklist.
- Arguments should match every device's enter schema conceptually.
- Desire calling vector_retrieve for conceptual native data.
- Desire calling web_search for latest or exterior data.
- Desire dataset_loader if the person asks a couple of named built-in dataset.
- Desire python_exec solely when computation or code execution is genuinely helpful.
- Don't fabricate unavailable instruments.
"""
immediate = f"""
USER TASK:
{activity}
EXPOSED TOOLS:
{json.dumps(exposed_tools, indent=2)}
Return JSON solely.
"""
obj = llm_json(directions, immediate)
raw_tool_calls = obj.get("tool_calls", [])
parsed_calls = []
allowed = {t["name"] for t in exposed_tools}
for name in raw_tool_calls[:MAX_TOOL_CALLS]:
title = name.get("tool_name", "")
args = name.get("arguments", {})
if title in allowed and isinstance(args, dict):
parsed_calls.append(ToolCall(tool_name=title, 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: Listing[ToolCall]) -> Listing[ToolResult]:
outcomes = []
for tc in tool_calls:
end result = self.server.tools_call(tc.tool_name, tc.arguments)
outcomes.append(end result)
return outcomes
def reply(self, activity: str, route: RouteDecision, exposed_tools: Listing[Dict[str, Any]], plan: PlanOutput, outcomes: Listing[ToolResult]) -> str:
directions = """
You're the last answering agent in an MCP-style routed device system.
Use the routed instruments and returned device outputs to reply the person.
Be concrete, concise, and technically right.
If device outputs are partial, say so.
Don't point out hidden instruments that weren't uncovered.
"""
tool_result_payload = [r.model_dump() for r in results]
immediate = f"""
USER TASK:
{activity}
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 reply the person clearly.
"""
resp = consumer.responses.create(
mannequin=self.mannequin,
enter=immediate,
directions=directions,
temperature=0.2
)
return resp.output_text
def run(self, activity: str, verbose: bool = True) -> Dict[str, Any]:
route = self.router.route(activity)
exposed_tools = self.discover_exposed_tools(route.selected_tools)
plan = self.plan(activity, exposed_tools)
outcomes = self.run_tools(plan.tool_calls) if plan.requires_tools else []
final_answer = self.reply(activity, route, exposed_tools, plan, outcomes)
payload = {
"activity": activity,
"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.match(f"USER TASKn{activity}", title="Enter"))
pretty_tools_table(exposed_tools, "Instruments Uncovered By MCP Router")
console.print(Panel(route.rationale or "No rationale supplied", title="Router Rationale"))
if route.policy_notes:
console.print(Panel("n".be a part of(f"- {x}" for x in route.policy_notes), title="Coverage Notes"))
console.print(Panel(plan.planner_note or "No planner observe supplied", title="Planner Observe"))
if outcomes:
for r in outcomes:
console.print(Panel.match(RichJSON.from_data(r.model_dump()), title=f"Software Consequence: {r.tool_name}"))
console.print(Panel(final_answer, title="Remaining Reply"))
return payload
def mcp_jsonrpc_tools_list(server: MCPToolServer) -> Dict[str, Any]:
return {
"jsonrpc": "2.0",
"id": 1,
"end result": {
"instruments": server.tools_list()
}
}
def mcp_jsonrpc_tools_call(server: MCPToolServer, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
end result = server.tools_call(tool_name, arguments)
return {
"jsonrpc": "2.0",
"id": 2,
"end result": end result.model_dump()
}
router = HybridMCPRouter(server=server, mannequin=MODEL)
agent = RoutedAgent(server=server, router=router, mannequin=MODEL)
console.print(Panel.match("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] after which clarify whether or not python execution was really obligatory.",
]
all_runs = []
for idx, activity in enumerate(demo_tasks, begin=1):
console.print(Panel.match(f"DEMO RUN {idx}", title="=" * 10))
out = agent.run(activity, verbose=True)
all_runs.append(out)
custom_task = "Design a routed MCP workflow for an AI analysis assistant that ought to use retrieval for native protocol data and net search solely when the duty explicitly asks for latest data."
custom_run = agent.run(custom_task, verbose=True)
print("nPROGRAMMATIC EXAMPLE: instruments/checklist")
print(json.dumps(mcp_jsonrpc_tools_list(server), indent=2))
print("nPROGRAMMATIC EXAMPLE: instruments/name for vector_retrieve")
print(json.dumps(mcp_jsonrpc_tools_call(server, "vector_retrieve", {"question": "dynamic functionality publicity in MCP routers", "top_k": 2}), indent=2))
print("nPROGRAMMATIC EXAMPLE: instruments/name for dataset_loader")
print(json.dumps(mcp_jsonrpc_tools_call(server, "dataset_loader", {"title": "iris", "n_rows": 5}), indent=2))
print("nPROGRAMMATIC EXAMPLE: customized last reply")
print(custom_run["final_answer"])