Construct an MCP Fashion Routed AI Agent System with Dynamic Software Publicity Planning, Execution, and Context Injection

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'll be able to 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 record.
- Arguments should match every device's enter schema conceptually.
- Choose calling vector_retrieve for conceptual native information.
- Choose calling web_search for latest or exterior data.
- Choose dataset_loader if the person asks a couple of named built-in dataset.
- Choose 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]:
           identify = name.get("tool_name", "")
           args = name.get("arguments", {})
           if identify in allowed and isinstance(args, dict):
               parsed_calls.append(ToolCall(tool_name=identify, 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:
           outcome = self.server.tools_call(tc.tool_name, tc.arguments)
           outcomes.append(outcome)
       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 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 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 = shopper.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 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 Word"))


           if outcomes:
               for r in outcomes:
                   console.print(Panel.match(RichJSON.from_data(r.model_dump()), title=f"Software Outcome: {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,
       "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 = [
   "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 essential.",
]


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 information and net search solely when the duty explicitly asks for latest data."
custom_run = agent.run(custom_task, verbose=True)


print("nPROGRAMMATIC EXAMPLE: instruments/record")
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", {"identify": "iris", "n_rows": 5}), indent=2))


print("nPROGRAMMATIC EXAMPLE: customized closing reply")
print(custom_run["final_answer"])