Una implementación para construir sistemas dinámicos de IA con el protocolo de contexto modelo (MCP) para la integración de recursos y herramientas en tiempo real

importar json importar asyncio desde clases de datos importar clase de datos, asdict desde escribir importar Dict, Lista, Cualquiera, Opcional, Invocable desde fecha y hora importar fecha y hora importar aleatorio clase @dataclass Recurso: uri: str nombre: str descripción: str mime_type: str contenido: Cualquiera = Ninguno @dataclass clase Herramienta: nombre: str descripción: str parámetros: Dict[str, Any]
manejador: Opcional[Callable] = Ninguno @dataclass clase Mensaje: rol: str contenido: str marca de tiempo: str = Ninguno def __post_init__(self): si no self.timestamp: self.timestamp = datetime.now().isoformat()

clase MCPServer: def __init__(self, nombre: str): self.name = nombre self.resources: Dict[str, Resource] = {} self.tools: Dict[str, Tool] = {} self.capabilities = {“resources”: True, “tools”: True, “prompts”: True, “logging”: True} print(f” ✓ MCP Server ‘{name}’ inicializado con capacidades: {list(self.capabilities.keys())}”) def Register_resource(self, recurso: Recurso) -> Ninguno: self.resources[resource.uri] = recurso print(f” → Recurso registrado: {recurso.nombre} ({recurso.uri})”) def registro_herramienta(self, herramienta: Herramienta) -> Ninguno: self.tools[tool.name] = herramienta print(f” → Herramienta registrada: {tool.name}”) async def get_resource(self, uri: str) -> Opcional[Resource]: await asyncio.sleep(0.1) return self.resources.get(uri) async def ejecutar_tool(self, nombre_herramienta: str, argumentos: Dict[str, Any]) -> Cualquiera: si nombre_herramienta no está en self.tools: aumente ValueError(f”Herramienta ‘{nombre_herramienta}’ no encontrada”) herramienta = self.tools[tool_name]
if tool.handler: return await tool.handler(**argumentos) return {“status”: “ejecutado”, “tool”: nombre_herramienta, “args”: argumentos} def list_resources(self) -> Lista[Dict[str, str]]: devolver [{“uri”: r.uri, “name”: r.name, “description”: r.description} for r in self.resources.values()]
def list_tools(self) -> Lista[Dict[str, Any]]: devolver [{“name”: t.name, “description”: t.description, “parameters”: t.parameters} for t in self.tools.values()]

clase MCPClient: def __init__(self, client_id: str): self.client_id = client_id self.connected_servers: Dict[str, MCPServer] = {} self.context: Lista[Message] = []
print(f”n ✓ Cliente MCP ‘{client_id}’ inicializado”) def connect_server(self, servidor: MCPServer) -> Ninguno: self.connected_servers[server.name] = servidor print(f” → Conectado al servidor: {server.name}”) async def query_resources(self, server_name: str) -> Lista[Dict[str, str]]: si nombre_servidor no está en self.connected_servers: aumente ValueError(f”No conectado al servidor: {server_name}”) devuelva self.connected_servers[server_name].list_resources() async def fetch_resource(self, server_name: str, uri: str) -> Opcional[Resource]: si nombre_servidor no está en self.connected_servers: aumente ValueError(f”No conectado al servidor: {server_name}”) servidor = self.connected_servers[server_name]
recurso = await server.get_resource(uri) if recurso: self.add_to_context(Message(role=”system”, content=f”Recurso obtenido: {resource.name}”)) return recurso async def call_tool(self, nombre_servidor: str, nombre_herramienta: str, **kwargs) -> Cualquiera: si nombre_servidor no está self.connected_servers: elevar ValueError(f”No conectado al servidor: {server_name}”) servidor = self.connected_servers[server_name]
resultado = await server.execute_tool(tool_name, kwargs) self.add_to_context(Message(role=”system”, content=f”Tool ‘{tool_name}’ ejecutado”)) devolver resultado def add_to_context(self, mensaje: Mensaje) -> Ninguno: self.context.append(message) def get_context(self) -> Lista[Dict[str, Any]]: devolver [asdict(msg) for msg in self.context]

async def analizar_sentimento (texto: str) -> Dict[str, Any]: await asyncio.sleep(0.2) sentimientos = [“positive”, “negative”, “neutral”]
return {“texto”: texto, “sentimiento”: aleatorio.choice(sentimientos), “confianza”: round(random.uniform(0.7, 0.99), 2)} async def resume_text(texto: str, max_length: int = 100) -> Dict[str, str]: espera asyncio.sleep(0.15) resumen = texto[:max_length] + “…” if len(texto) > max_length else text return {“original_length”: len(texto), “summary”: resumen, “compression_ratio”: round(len(summary) / len(texto), 2)} async def search_knowledge(consulta: str, top_k: int = 3) -> Lista[Dict[str, Any]]: espera asyncio.sleep(0.25) simulacro_resultados = [{“title”: f”Result {i+1} for ‘{query}'”, “score”: round(random.uniform(0.5, 1.0), 2)} for i in range(top_k)]
retorno ordenado (mock_results, clave = lambda x: x[“score”]reverso=Verdadero)

async def run_mcp_demo(): print(“=” * 60) print(“PROTOCOLO DE CONTEXTO DEL MODELO (MCP) – TUTORIAL AVANZADO”) print(“=” * 60) print(“n[1] Configurando el servidor MCP…”) servidor = MCPServer(“servidor de conocimientos”) print(“n[2] Registrando recursos…”) server.register_resource(Resource(uri=”docs://python-guide”, name=”Guía de programación de Python”, descripción=”Documentación completa de Python”, mime_type=”text/markdown”, content=”# Guía de PythonPython es un lenguaje de programación de alto nivel…”)) server.register_resource(Resource(uri=”data://sales-2024″, name=”2024 Sales Data”, descripción=”Métricas de ventas anuales”, mime_type=”application/json”, content={“q1”: 125000, “q2”: 142000, “q3”: 138000, “q4”: 165000})) print(“n[3] Registrando herramientas…”) server.register_tool(Tool(name=”analyze_sentiment”, descripción=”Analizar sentimiento de texto”, parámetros={“text”: {“type”: “string”, “required”: True}}, handler=analyze_sentiment)) server.register_tool(Tool(name=”summarize_text”, descripción=”Resumir texto largo”, parámetros={“text”: {“type”: “string”, “required”: True}, “max_length”: {“type”: “integer”, “default”: 100}}, handler=summarize_text)) server.register_tool(Tool(name=”search_knowledge”, descripción=”Buscar base de conocimientos”, parámetros={“query”: {“type”: “string”, “required”: True}, “top_k”: {“type”: “integer”, “default”: 3}}, handler=search_knowledge)) cliente = MCPClient(“demo-client”) client.connect_server(servidor) print(“n” + “=” * 60) print(“DEMOSTRACIÓN: MCP EN ACCIÓN”) print(“=” * 60) print(“n[Demo 1] Listado de recursos disponibles…”) recursos = await client.query_resources(“knowledge-server”) para res en recursos: print(f” • {res[‘name’]}: {resolución[‘description’]}”) imprimir(“n[Demo 2] Obteniendo recurso de datos de ventas…”) sales_resource = await client.fetch_resource(“knowledge-server”, “data://sales-2024″) if sales_resource: print(f” Datos: {json.dumps(sales_resource.content, indent=2)}”) print(“n[Demo 3] Analizando sentimiento…”) sentiment_result = await client.call_tool(“knowledge-server”, “analyze_sentiment”, text=”¡MCP es un protocolo increíble para la integración de IA!”) print(f” Resultado: {json.dumps(sentiment_result, indent=2)}”) print(“n[Demo 4] Texto resumido…”) resumen_result = await client.call_tool(“knowledge-server”, “summarize_text”, text=”El protocolo de contexto del modelo permite una integración perfecta entre modelos de IA y fuentes de datos externas…”, max_length=50) print(f” Resumen: {summary_result[‘summary’]}”) imprimir(“n[Demo 5] Buscando base de conocimientos…”) search_result = await client.call_tool(“knowledge-server”, “search_knowledge”, query=”machine learning”, top_k=3) print(” Principales resultados:”) para obtener resultados en search_result: print(f” – {resultado[‘title’]} (puntuación: {resultado[‘score’]})”) imprimir(“n[Demo 6] Ventana de contexto actual…”) contexto = client.get_context() print(f” Longitud del contexto: {len(context)} mensajes”) para i, msg en enumerate(contexto[-3:]1): imprimir(f” {i}. [{msg[‘role’]}]{mensaje[‘content’]}”) print(“n” + “=” * 60) print(“¡Tutorial de MCP completo!”) print(“=” * 60) print(“nConclusiones clave:”) print(“• MCP permite conexiones modulares de IA a recursos”) print(“• Los recursos proporcionan contexto de fuentes externas”) print(“• Las herramientas permiten operaciones y acciones dinámicas”) print(“• El diseño asíncrono admite operaciones de E/S eficientes”) if __name__ == “__main__”: importe sys si ‘ipykernel’ en sys.modules o ‘google.colab’ en sys.modules: espere run_mcp_demo() más: asyncio.run(run_mcp_demo())