Build a Dynamic MCP-Style AI Agent System with Adaptive Tool Routing, Execution, and Context Injection

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 operating within an MCP-like framework.
You are limited to using only the tools that have been made available.
Determine whether any tools are required.
Respond exclusively in JSON format with the following keys:
requires_tools: boolean
tool_calls: array of objects containing tool_name and arguments
direct_answer_allowed: boolean
planner_note: string


Guidelines:
- Limit yourself to a maximum of 3 tool calls.
- Only invoke tools from the provided list.
- Ensure arguments align with each tool's expected input structure.
- Use vector_retrieve for conceptual or internal knowledge lookups.
- Use web_search for up-to-date or external information.
- Use dataset_loader when the user references a specific built-in dataset.
- Reserve python_exec for cases where actual computation or code execution is beneficial.
- Never invent tools that aren't available.
"""


       prompt = f"""
USER TASK:
{task}


AVAILABLE TOOLS:
{json.dumps(exposed_tools, indent=2)}


Provide JSON output 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 serve as the final response agent in an MCP-style routed tool environment.
Leverage the routed tools and their outputs to craft your reply.
Be specific, brief, and technically accurate.
Acknowledge if tool outputs are incomplete.
Avoid referencing any tools that were not made available.
"""


       tool_result_payload = [r.model_dump() for r in results]


       prompt = f"""
USER TASK:
{task}


ROUTING DECISION:
{route.model_dump_json(indent=2)}


AVAILABLE TOOLS:
{json.dumps(exposed_tools, indent=2)}


PLAN:
{plan.model_dump_json(indent=2)}


TOOL OUTPUTS:
{json.dumps(tool_result_payload, indent=2)}


Now provide a clear response to the user.
"""
       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 TASKn{task}", title="Input"))
           pretty_tools_table(exposed_tools, "Tools Made Available 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 Output: {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 = [
   "Describe how an MCP tool router should reveal tools for an agent task involving dynamic capability exposure.",
   "Search the internet for recent MCP-related developments and provide a summary.",
   "Load the iris dataset, examine its columns and basic statistics, and identify the type of ML problem it represents.",
   "Retrieve internal knowledge about context injection and router policies, then explain how limiting tool access improves agent performance.",
   "Use Python to calculate the average of [3, 5, 9, 10, 13] and then assess whether using Python 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 uses retrieval for internal protocol knowledge and web search only when the task specifically requests 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"])