class RoutedAgent:
\n def __init__(self, server: MCPToolServer, router: HybridMCPRouter, model: str):
\n self.server = server
\n self.router = router
\n self.model = model<\/p>\n
def discover_exposed_tools(self, exposed_tool_names: List[str]) -> List[Dict[str, Any]]:
\n return [t for t in self.server.tools_list() if t[“name”] in exposed_tool_names]<\/p>\n
def plan(self, task: str, exposed_tools: List[Dict[str, Any]]) -> PlanOutput:
\n instructions = “””
\nYou are a planning agent in an MCP-like architecture.
\nYou can only use the exposed tools.
\nDecide whether tools are needed.
\nReturn strict JSON only with keys:
\nrequires_tools: boolean
\ntool_calls: array of objects with tool_name and arguments
\ndirect_answer_allowed: boolean
\nplanner_note: string<\/p>\n
Rules:
\n– Use at most 3 tool calls.
\n– Only call tools from the exposed list.
\n– Arguments must match each tool’s input schema conceptually.
\n– Prefer calling vector_retrieve for conceptual local knowledge.
\n– Prefer calling web_search for recent or external information.
\n– Prefer dataset_loader if the user asks about a named built-in dataset.
\n– Prefer python_exec only when computation or code execution is genuinely useful.
\n– Do not fabricate unavailable tools.
\n“””<\/p>\n
prompt = f”””
\nUSER TASK:
\n{task}<\/p>\n
EXPOSED TOOLS:
\n{json.dumps(exposed_tools, indent=2)}<\/p>\n
Return JSON only.
\n“””
\n obj = llm_json(instructions, prompt)<\/p>\n
raw_tool_calls = obj.get(“tool_calls”, [])
\n parsed_calls = []
\n allowed = {t[“name”] for t in exposed_tools}<\/p>\n
for call in raw_tool_calls[:MAX_TOOL_CALLS]:
\n name = call.get(“tool_name”, “”)
\n args = call.get(“arguments”, {})
\n if name in allowed and isinstance(args, dict):
\n parsed_calls.append(ToolCall(tool_name=name, arguments=args))<\/p>\n
return PlanOutput(
\n requires_tools=bool(obj.get(“requires_tools”, False) or parsed_calls),
\n tool_calls=parsed_calls,
\n direct_answer_allowed=bool(obj.get(“direct_answer_allowed”, False)),
\n planner_note=obj.get(“planner_note”, “”),
\n )<\/p>\n
def run_tools(self, tool_calls: List[ToolCall]) -> List[ToolResult]:
\n results = []
\n for tc in tool_calls:
\n result = self.server.tools_call(tc.tool_name, tc.arguments)
\n results.append(result)
\n return results<\/p>\n
def answer(self, task: str, route: RouteDecision, exposed_tools: List[Dict[str, Any]], plan: PlanOutput, results: List[ToolResult]) -> str:
\n instructions = “””
\nYou are the final answering agent in an MCP-style routed tool system.
\nUse the routed tools and returned tool outputs to answer the user.
\nBe concrete, concise, and technically correct.
\nIf tool outputs are partial, say so.
\nDo not mention hidden tools that were not exposed.
\n“””<\/p>\n
tool_result_payload = [r.model_dump() for r in results]<\/p>\n
prompt = f”””
\nUSER TASK:
\n{task}<\/p>\n
ROUTE DECISION:
\n{route.model_dump_json(indent=2)}<\/p>\n
EXPOSED TOOLS:
\n{json.dumps(exposed_tools, indent=2)}<\/p>\n
PLAN:
\n{plan.model_dump_json(indent=2)}<\/p>\n
TOOL RESULTS:
\n{json.dumps(tool_result_payload, indent=2)}<\/p>\n
Now answer the user clearly.
\n“””
\n resp = client.responses.create(
\n model=self.model,
\n input=prompt,
\n instructions=instructions,
\n temperature=0.2
\n )
\n return resp.output_text<\/p>\n
def run(self, task: str, verbose: bool = True) -> Dict[str, Any]:
\n route = self.router.route(task)
\n exposed_tools = self.discover_exposed_tools(route.selected_tools)
\n plan = self.plan(task, exposed_tools)
\n results = self.run_tools(plan.tool_calls) if plan.requires_tools else []
\n final_answer = self.answer(task, route, exposed_tools, plan, results)<\/p>\n
payload = {
\n “task”: task,
\n “route_decision”: route.model_dump(),
\n “exposed_tools”: exposed_tools,
\n “plan”: plan.model_dump(),
\n “tool_results”: [r.model_dump() for r in results],
\n “final_answer”: final_answer,
\n }<\/p>\n
if verbose:
\n console.print(Panel.fit(f”USER TASK\\n{task}”, title=”Input”))
\n pretty_tools_table(exposed_tools, “Tools Exposed By MCP Router”)
\n console.print(Panel(route.rationale or “No rationale provided”, title=”Router Rationale”))
\n if route.policy_notes:
\n console.print(Panel(“\\n”.join(f”- {x}” for x in route.policy_notes), title=”Policy Notes”))
\n console.print(Panel(plan.planner_note or “No planner note provided”, title=”Planner Note”))<\/p>\n
if results:
\n for r in results:
\n console.print(Panel.fit(RichJSON.from_data(r.model_dump()), title=f”Tool Result: {r.tool_name}”))
\n console.print(Panel(final_answer, title=”Final Answer”))<\/p>\n
return payload<\/p>\n
def mcp_jsonrpc_tools_list(server: MCPToolServer) -> Dict[str, Any]:
\n return {
\n “jsonrpc”: “2.0”,
\n “id”: 1,
\n “result”: {
\n “tools”: server.tools_list()
\n }
\n }<\/p>\n
def mcp_jsonrpc_tools_call(server: MCPToolServer, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
\n result = server.tools_call(tool_name, arguments)
\n return {
\n “jsonrpc”: “2.0”,
\n “id”: 2,
\n “result”: result.model_dump()
\n }<\/p>\n
router = HybridMCPRouter(server=server, model=MODEL)
\nagent = RoutedAgent(server=server, router=router, model=MODEL)<\/p>\n
console.print(Panel.fit(“MCP-STYLE TOOL DISCOVERY”, title=”Step 1″))
\nconsole.print(RichJSON.from_data(mcp_jsonrpc_tools_list(server)))<\/p>\n
demo_tasks = [
\n “Explain how an MCP tool router should expose tools for an agent task about dynamic capability exposure.”,
\n “Search the web for recent examples of MCP-related developments and summarize them.”,
\n “Load the iris dataset, inspect its columns and basic stats, and tell me what kind of ML problem it is.”,
\n “Retrieve local knowledge about context injection and router policies, then explain why restricting tool access helps agent performance.”,
\n “Use Python to compute the average of [3, 5, 9, 10, 13] and then explain whether python execution was truly necessary.”,
\n]<\/p>\n
all_runs = []
\nfor idx, task in enumerate(demo_tasks, start=1):
\n console.print(Panel.fit(f”DEMO RUN {idx}”, title=”=” * 10))
\n out = agent.run(task, verbose=True)
\n all_runs.append(out)<\/p>\n
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.”
\ncustom_run = agent.run(custom_task, verbose=True)<\/p>\n
print(“\\nPROGRAMMATIC EXAMPLE: tools\/list”)
\nprint(json.dumps(mcp_jsonrpc_tools_list(server), indent=2))<\/p>\n
print(“\\nPROGRAMMATIC EXAMPLE: tools\/call for vector_retrieve”)
\nprint(json.dumps(mcp_jsonrpc_tools_call(server, “vector_retrieve”, {“query”: “dynamic capability exposure in MCP routers”, “top_k”: 2}), indent=2))<\/p>\n
print(“\\nPROGRAMMATIC EXAMPLE: tools\/call for dataset_loader”)
\nprint(json.dumps(mcp_jsonrpc_tools_call(server, “dataset_loader”, {“name”: “iris”, “n_rows”: 5}), indent=2))<\/p>\n
print(“\\nPROGRAMMATIC EXAMPLE: custom final answer”)
\nprint(custom_run[“final_answer”])<\/p>\n","protected":false},"excerpt":{"rendered":"class RoutedAgent: def __init__(self, server: MCPToolServer, router: HybridMCPRouter, model: str): self.server = server self.router = router self.model =…\n","protected":false},"author":2,"featured_media":675751,"comment_status":"","ping_status":"","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[20],"tags":[256,254,255,64,63,105],"class_list":["post-675750","post","type-post","status-publish","format-standard","has-post-thumbnail","category-artificial-intelligence","tag-ai","tag-artificial-intelligence","tag-artificialintelligence","tag-au","tag-australia","tag-technology"],"_links":{"self":[{"href":"https:\/\/www.newsbeep.com\/au\/wp-json\/wp\/v2\/posts\/675750","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.newsbeep.com\/au\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.newsbeep.com\/au\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.newsbeep.com\/au\/wp-json\/wp\/v2\/users\/2"}],"replies":[{"embeddable":true,"href":"https:\/\/www.newsbeep.com\/au\/wp-json\/wp\/v2\/comments?post=675750"}],"version-history":[{"count":0,"href":"https:\/\/www.newsbeep.com\/au\/wp-json\/wp\/v2\/posts\/675750\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/www.newsbeep.com\/au\/wp-json\/wp\/v2\/media\/675751"}],"wp:attachment":[{"href":"https:\/\/www.newsbeep.com\/au\/wp-json\/wp\/v2\/media?parent=675750"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.newsbeep.com\/au\/wp-json\/wp\/v2\/categories?post=675750"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.newsbeep.com\/au\/wp-json\/wp\/v2\/tags?post=675750"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}