MCP HTTP TAVILY DATE OAUTH

import asyncio import json import sys import requests from datetime import datetime from typing import Dict, Any, Optional # Cliente HTTP para servidor MCP # Prueba el servidor MCP HTTP en lugar de STDIO class MCPHTTPClient: """Cliente HTTP para servidor MCP""" def __init__(self, base_url: str = None): # Detectar URL automáticamente si no se proporciona if base_url is None: import os if os.getenv("PORT"): # Railway base_url = "https://mcptavilidateoaut-production.up.railway.app" else: # Desarrollo local base_url = "http://localhost:8001" self.base_url = base_url self.mcp_url = f"{base_url}/mcp" self.session = requests.Session() self.request_id = 0 def _get_next_id(self) -> str: """Genera el siguiente ID de petición""" self.request_id += 1 return str(self.request_id) def _make_mcp_request(self, method: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """Realiza una petición MCP HTTP""" request_data = { "jsonrpc": "2.0", "id": self._get_next_id(), "method": method } if params: request_data["params"] = params print(f"Enviando petición MCP: {method}") print(f"Datos: {json.dumps(request_data, indent=2, ensure_ascii=False)}") try: response = self.session.post( self.mcp_url, json=request_data, headers={"Content-Type": "application/json"}, timeout=30 ) response.raise_for_status() result = response.json() print(f"Respuesta recibida:") print(f"{json.dumps(result, indent=2, ensure_ascii=False)}") return result except requests.exceptions.RequestException as e: print(f"Error en petición HTTP: {e}") return {"error": str(e)} except json.JSONDecodeError as e: print(f"Error decodificando JSON: {e}") return {"error": f"JSON decode error: {e}"} def initialize(self) -> Dict[str, Any]: """Inicializa el servidor MCP""" params = { "protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": { "name": "mcp-http-client", "version": "1.0.0" } } return self._make_mcp_request("initialize", params) def initialized(self) -> Dict[str, Any]: """Envía notificación de inicialización completada""" request_data = { "jsonrpc": "2.0", "method": "notifications/initialized", "params": {} } print(f"Enviando notificación: initialized") print(f"Datos: {json.dumps(request_data, indent=2, ensure_ascii=False)}") try: response = self.session.post( self.mcp_url, json=request_data, headers={"Content-Type": "application/json"}, timeout=30 ) response.raise_for_status() # Las notificaciones pueden no devolver respuesta if response.content: result = response.json() print(f"Respuesta recibida:") print(f"{json.dumps(result, indent=2, ensure_ascii=False)}") return result else: print("Notificación enviada (sin respuesta)") return {"status": "sent"} except requests.exceptions.RequestException as e: print(f"Error en notificación HTTP: {e}") return {"error": str(e)} def list_tools(self) -> Dict[str, Any]: """Lista las herramientas disponibles""" return self._make_mcp_request("tools/list") def call_tool(self, name: str, arguments: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """Llama a una herramienta específica""" params = { "name": name, "arguments": arguments or {} } return self._make_mcp_request("tools/call", params) def health_check(self) -> Dict[str, Any]: """Verifica el estado del servidor""" try: response = self.session.get(f"{self.base_url}/health", timeout=10) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: return {"error": str(e)} def get_server_info(self) -> Dict[str, Any]: """Obtiene información del servidor""" try: response = self.session.get(f"{self.base_url}/", timeout=10) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: return {"error": str(e)} async def test_mcp_http_server(): """Prueba completa del servidor MCP HTTP""" print("=" * 60) print("PRUEBA DEL SERVIDOR MCP HTTP") print("=" * 60) client = MCPHTTPClient() # 1. Verificar que el servidor esté funcionando print("\n1. Verificando estado del servidor...") health = client.health_check() if "error" in health: print(f"ERROR - Error conectando al servidor: {health['error']}") print("Asegúrate de que el servidor esté ejecutándose correctamente") return else: print(f"OK - Servidor funcionando: {health}") # 2. Obtener información del servidor print("\n2. Información del servidor...") server_info = client.get_server_info() if "error" not in server_info: print(f"OK - Servidor: {server_info['name']} v{server_info['version']}") print(f" Herramientas: {', '.join(server_info['tools'])}") else: print(f"ERROR - Error obteniendo información: {server_info['error']}") # 3. Inicializar MCP print("\n3. Inicializando protocolo MCP...") init_result = client.initialize() if "error" in init_result: print(f"ERROR - Error en inicialización: {init_result['error']}") return else: print("OK - MCP inicializado correctamente") server_info = init_result.get("result", {}).get("serverInfo", {}) print(f" Servidor: {server_info.get('name')} v{server_info.get('version')}") # 4. Enviar notificación de inicialización completada print("\n4. Enviando notificación de inicialización...") init_notif = client.initialized() print("OK - Notificación enviada") # 5. Listar herramientas print("\n5. Listando herramientas disponibles...") tools_result = client.list_tools() if "error" in tools_result: print(f"❌ Error listando herramientas: {tools_result['error']}") return else: tools = tools_result.get("result", {}).get("tools", []) print(f"✅ Herramientas disponibles ({len(tools)}):") for tool in tools: print(f" - {tool['name']}: {tool['description']}") # 6. Probar herramienta datetime_now print("\n6. Probando herramienta datetime_now...") datetime_result = client.call_tool("datetime_now") if "error" in datetime_result: print(f"❌ Error en datetime_now: {datetime_result['error']}") else: content = datetime_result.get("result", {}).get("content", []) if content and len(content) > 0: datetime_text = content[0].get("text", "") print(f"✅ Fecha/hora actual: {datetime_text}") else: print("❌ No se recibió contenido en la respuesta") # 7. Probar herramienta web_search print("\n7. Probando herramienta web_search...") web_search_result = client.call_tool( "web_search", { "query": "What is Ollama and how does it work?", "model": "gpt-oss:20b" } ) if "error" in web_search_result: print(f"❌ Error en web_search: {web_search_result['error']}") else: content = web_search_result.get("result", {}).get("content", []) if content and len(content) > 0: web_text = content[0].get("text", "") print(f"✅ Búsqueda web completada:") print(f" Resultado ({len(web_text)} caracteres):") # Mostrar solo los primeros 200 caracteres preview = web_text[:200] + "..." if len(web_text) > 200 else web_text print(f" {preview}") else: print("❌ No se recibió contenido en la respuesta") print("\n" + "=" * 60) print("PRUEBA COMPLETADA") print("=" * 60) def test_direct_endpoints(): """Prueba los endpoints directos (sin MCP)""" print("\n" + "=" * 60) print("PRUEBA DE ENDPOINTS DIRECTOS") print("=" * 60) # Detectar URL automáticamente import os if os.getenv("PORT"): # Railway base_url = "https://mcptavilidateoaut-production.up.railway.app" else: # Desarrollo local base_url = "http://localhost:8001" # Probar endpoint de salud print("\n1. Probando endpoint /health...") try: response = requests.get(f"{base_url}/health", timeout=10) response.raise_for_status() health_data = response.json() print(f"✅ Health check: {health_data}") except Exception as e: print(f"❌ Error en health check: {e}") # Probar endpoint de herramientas print("\n2. Probando endpoint /tools...") try: response = requests.get(f"{base_url}/tools", timeout=10) response.raise_for_status() tools_data = response.json() print(f"✅ Tools endpoint: {tools_data}") except Exception as e: print(f"❌ Error en tools endpoint: {e}") # Probar endpoint raíz print("\n3. Probando endpoint raíz...") try: response = requests.get(f"{base_url}/", timeout=10) response.raise_for_status() root_data = response.json() print(f"✅ Root endpoint: {root_data}") except Exception as e: print(f"❌ Error en root endpoint: {e}") if __name__ == "__main__": print("Cliente HTTP para servidor MCP") print("Asegúrate de que el servidor esté ejecutándose:") print(" uv run python src/server.py") print() # Ejecutar pruebas asyncio.run(test_mcp_http_server()) test_direct_endpoints()