Cómo construir un sistema de agentes de IA enrutados estilo MCP con planificación, ejecución e inyección de contexto de exposición de herramientas dinámicas

clase RoutedAgent: def __init__(self, servidor: MCPToolServer, enrutador: HybridMCPRouter, modelo: str): self.server = servidor self.router = enrutador self.model = modelo def discover_exposed_tools(self, nombres_herramientas_expuestas: Lista[str]) -> Lista[Dict[str, Any]]: devolver [t for t in self.server.tools_list() if t[“name”] en nombres_herramientas_expuestas]def plan(self, tarea: str, herramientas_expuestas: Lista[Dict[str, Any]]) -> PlanOutput: instrucciones = “”” Usted es un agente de planificación en una arquitectura tipo MCP. Solo puede usar las herramientas expuestas. Decida si se necesitan herramientas. Devuelve JSON estricto solo con claves: require_tools: boolean tool_calls: matriz de objetos con nombre_herramienta y argumentos direct_answer_allowed: boolean Planner_note: cadena Reglas: – Use como máximo 3 llamadas a herramientas. – Solo llame a herramientas de la lista expuesta. – Los argumentos deben coincidir el esquema de entrada de cada herramienta conceptualmente. – Prefiere llamar a vector_retrieve para obtener conocimiento local conceptual. – Prefiere llamar a web_search para obtener información reciente o externa. – Prefiere dataset_loader si el usuario pregunta sobre un conjunto de datos integrado con nombre. {json.dumps(exposed_tools, indent=2)} Devuelve JSON solamente “”” obj = llm_json(instrucciones, aviso) raw_tool_calls = obj.get(“tool_calls”, []) llamadas_analizadas = []
permitido = {t[“name”] para t en expuesto_tools} para llamada en raw_tool_calls[:MAX_TOOL_CALLS]: nombre = call.get(“tool_name”, “”) args = call.get(“arguments”, {}) si el nombre está permitido y isinstance(args, dict): parsed_calls.append(ToolCall(tool_name=name, arguments=args)) return PlanOutput( require_tools=bool(obj.get(“requires_tools”, False) o 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: Lista[ToolCall]) -> Lista[ToolResult]: resultados = []
para tc en llamadas_herramientas: resultado = self.server.tools_call(tc.tool_name, tc.arguments) resultados.append(resultado) devuelve resultados def respuesta(self, tarea: cadena, ruta: RouteDecision, herramientas_expuestas: Lista[Dict[str, Any]], plan: PlanOutput, resultados: Lista[ToolResult]) -> str: instrucciones = “”” Usted es el agente de respuesta final en un sistema de herramientas enrutadas estilo MCP. Utilice las herramientas enrutadas y los resultados de las herramientas devueltas para responder al usuario. Sea concreto, conciso y técnicamente correcto. Si los resultados de las herramientas son parciales, dígalo. No mencione herramientas ocultas que no estuvieron expuestas. “”” tool_result_payload = [r.model_dump() for r in results]

Prompt = f””” TAREA DEL USUARIO: {task} DECISIÓN DE RUTA: {route.model_dump_json(indent=2)} HERRAMIENTAS EXPUESTAS: {json.dumps(exposed_tools, indent=2)} PLAN: {plan.model_dump_json(indent=2)} RESULTADOS DE LA HERRAMIENTA: {json.dumps(tool_result_payload, indent=2)} Ahora responda el usuario claramente “”” resp = client.responses.create( model=self.model, input=prompt, instrucciones=instrucciones, temperatura=0.2 ) return resp.output_text def run(self, task: str, detallado: bool = True) -> Dict.[str, Any]: ruta = self.router.route(tarea) herramientas_expuestas = self.discover_exposed_tools(ruta.selected_tools) plan = self.plan(tarea, herramientas_expuestas) resultados = self.run_tools(plan.tool_calls) si plan.requires_tools else []
respuesta_final = self.answer(tarea, ruta, herramientas_expuestas, plan, resultados) carga útil = { “tarea”: ​​tarea, “ruta_decisión”: ruta.model_dump(), “herramientas_expuestas”: herramientas_expuestas, “plan”: plan.model_dump(), “resultados_herramientas”: [r.model_dump() for r in results]”final_answer”: final_answer, } si es detallado: console.print(Panel.fit(f”USER TASK\n{task}”, title=”Input”)) Pretty_tools_table(exposed_tools, “Herramientas expuestas por el enrutador MCP”) console.print(Panel(route.rationale o “No se proporciona ninguna justificación”, título=”Router Justificación”)) si route.policy_notes: console.print(Panel(“\n”.join(f”- {x}” for x in route.policy_notes), title=”Notas de política”)) console.print(Panel(plan.planner_note or “No se proporcionó nota del planificador”, título=”Nota del planificador”)) si resultados: para r en resultados: console.print(Panel.fit(RichJSON.from_data(r.model_dump()), title=f”Resultado de la herramienta: {r.tool_name}”)) console.print(Panel(final_answer, title=”Respuesta final”)) devolver carga útil def mcp_jsonrpc_tools_list(servidor: MCPToolServer) -> Dict[str, Any]: return { “jsonrpc”: “2.0”, “id”: 1, “resultado”: { “tools”: server.tools_list() } } def mcp_jsonrpc_tools_call(servidor: MCPToolServer, nombre_herramienta: str, argumentos: Dict[str, Any]) -> Dictar[str, Any]: resultado = server.tools_call(nombre_herramienta, argumentos) return { “jsonrpc”: “2.0”, “id”: 2, “resultado”: resultado.model_dump() } enrutador = HybridMCPRouter(servidor=servidor, modelo=MODELO) agente = RoutedAgent(servidor=servidor, enrutador=enrutador, modelo=MODELO) console.print(Panel.fit(“DESCUBRIMIENTO DE HERRAMIENTA ESTILO MCP”, title=”Paso 1″)) console.print(RichJSON.from_data(mcp_jsonrpc_tools_list(servidor))) 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] y luego explique si la ejecución de Python fue realmente necesaria.”, ]all_runs = []
para idx, tarea en 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 = “Diseñe un flujo de trabajo MCP enrutado para un asistente de investigación de IA que debería utilizar la recuperación de conocimiento del protocolo local y búsqueda web solo cuando la tarea solicite explícitamente información reciente información.” custom_run = agent.run(custom_task, verbose=True) print(“\nEJEMPLO PROGRAMÁTICO: herramientas/lista”) print(json.dumps(mcp_jsonrpc_tools_list(server), indent=2)) print(“\nEJEMPLO PROGRAMÁTICO: herramientas/llamada para vector_retrieve”) print(json.dumps(mcp_jsonrpc_tools_call(server, “vector_retrieve”, {“query”: “exposición de capacidad dinámica en enrutadores MCP”, “top_k”: 2}), indent=2)) print(“\nEJEMPLO DE PROGRAMACIÓN: herramientas/llamada para dataset_loader”) print(json.dumps(mcp_jsonrpc_tools_call(server, “dataset_loader”, {“name”: “iris”, “n_rows”: 5}), indent=2)) print(“\nEJEMPLO PROGRAMÁTICO: respuesta final personalizada”) print(custom_run[“final_answer”])