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: Record(str)) -> Record(Dict(str, Any)):
return (t for t in self.server.tools_list() if t("title") in exposed_tool_names)
def plan(self, activity: str, exposed_tools: Record(Dict(str, Any))) -> PlanOutput:
directions = """
You're a planning agent in an MCP-like structure.
You possibly can solely use the uncovered instruments.
Resolve 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 listing.
- 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 info.
- 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("title") 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: Record(ToolCall)) -> Record(ToolResult):
outcomes = ()
for tc in tool_calls:
outcome = self.server.tools_call(tc.tool_name, tc.arguments)
outcomes.append(outcome)
return outcomes
def reply(self, activity: str, route: RouteDecision, exposed_tools: Record(Dict(str, Any)), plan: PlanOutput, outcomes: Record(ToolResult)) -> str:
directions = """
You're the closing 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 appropriate.
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 outcomes)
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 outcomes),
"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 word supplied", title="Planner Observe"))
if outcomes:
for r in outcomes:
console.print(Panel.match(RichJSON.from_data(r.model_dump()), title=f"Software End result: {r.tool_name}"))
console.print(Panel(final_answer, title="Last Reply"))
return payload
def mcp_jsonrpc_tools_list(server: MCPToolServer) -> Dict(str, Any):
return {
"jsonrpc": "2.0",
"id": 1,
"outcome": {
"instruments": server.tools_list()
}
}
def mcp_jsonrpc_tools_call(server: MCPToolServer, tool_name: str, arguments: Dict(str, Any)) -> Dict(str, Any):
outcome = server.tools_call(tool_name, arguments)
return {
"jsonrpc": "2.0",
"id": 2,
"outcome": outcome.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 = (
"Clarify how an MCP device router ought to expose instruments for an agent activity about dynamic functionality publicity.",
"Search the online for latest examples of MCP-related developments and summarize them.",
"Load the iris dataset, examine its columns and primary stats, and inform me what sort of ML drawback it's.",
"Retrieve native data about context injection and router insurance policies, then clarify why proscribing device entry helps agent efficiency.",
"Use Python to compute the typical of (3, 5, 9, 10, 13) after which clarify whether or not python execution was actually crucial.",
)
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 info."
custom_run = agent.run(custom_task, verbose=True)
print("nPROGRAMMATIC EXAMPLE: instruments/listing")
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 closing reply")
print(custom_run("final_answer"))