IRIS Legacy

""" Generic MCP Client - Universal JSON-RPC 2.0 client for any MCP server This client can communicate with any MCP-compliant server using the standard protocol: - tools/list - List available tools - tools/call - Execute a tool Supports both IRIS MCP Server (localhost) and external MCP servers (Infocert, etc.) """ import logging import aiohttp import json from typing import Dict, Any, List, Optional from dataclasses import dataclass logger = logging.getLogger(__name__) @dataclass class MCPServer: """Configuration for an MCP server""" name: str url: str api_key: Optional[str] = None description: Optional[str] = None class MCPError(Exception): """MCP protocol error""" def __init__(self, message: str, code: int = -1, data: Any = None): super().__init__(message) self.code = code self.data = data class GenericMCPClient: """ Universal MCP client using JSON-RPC 2.0 protocol. Can communicate with any MCP server that implements: - tools/list: List available tools - tools/call: Execute a tool with parameters Example usage: client = GenericMCPClient() # Add MCP servers client.add_server(MCPServer( name="iris", url="http://localhost:8001/mcp/" )) client.add_server(MCPServer( name="infocert", url="https://mcp.uat.brainaihub.tech/digital-signature/sse", api_key="your-api-key" )) # List tools from a server tools = await client.list_tools("iris") # Call a tool result = await client.call_tool( "iris", "email_list_messages", {"user_email": "user@example.com"} ) """ def __init__(self, timeout: int = 30): """ Initialize MCP client. Args: timeout: Request timeout in seconds """ self.servers: Dict[str, MCPServer] = {} self.timeout = aiohttp.ClientTimeout(total=timeout) self._request_id = 0 logger.info("Generic MCP Client initialized") def add_server(self, server: MCPServer): """ Register an MCP server. Args: server: MCP server configuration """ self.servers[server.name] = server logger.info(f"Added MCP server: {server.name} ({server.url})") def _get_next_id(self) -> int: """Generate next JSON-RPC request ID""" self._request_id += 1 return self._request_id async def _jsonrpc_call( self, server_name: str, method: str, params: Optional[Dict[str, Any]] = None ) -> Any: """ Make a JSON-RPC 2.0 call to an MCP server. Args: server_name: Name of the registered MCP server method: JSON-RPC method (e.g., "tools/list", "tools/call") params: Method parameters Returns: Result from the JSON-RPC response Raises: MCPError: If server not found or JSON-RPC error occurs """ if server_name not in self.servers: raise MCPError(f"MCP server not found: {server_name}") server = self.servers[server_name] # Build JSON-RPC 2.0 request request = { "jsonrpc": "2.0", "id": self._get_next_id(), "method": method, "params": params or {} } # Build headers headers = { "Content-Type": "application/json" } if server.api_key: headers["Authorization"] = f"Bearer {server.api_key}" logger.debug(f"MCP Request to {server.name}: {method}") logger.debug(f"Params: {json.dumps(params, indent=2)}") try: async with aiohttp.ClientSession(timeout=self.timeout) as session: async with session.post( server.url, json=request, headers=headers ) as response: response_text = await response.text() logger.debug(f"MCP Response ({response.status}): {response_text[:500]}") # Parse JSON-RPC response try: response_data = json.loads(response_text) except json.JSONDecodeError as e: raise MCPError( f"Invalid JSON response from {server.name}: {str(e)}", code=-32700 ) # Check for JSON-RPC error if "error" in response_data: error = response_data["error"] raise MCPError( error.get("message", "Unknown error"), code=error.get("code", -1), data=error.get("data") ) # Return result if "result" not in response_data: raise MCPError( "Invalid JSON-RPC response: missing 'result'", code=-32600 ) return response_data["result"] except aiohttp.ClientError as e: logger.error(f"HTTP error calling {server.name}: {e}") raise MCPError(f"Connection error to {server.name}: {str(e)}") except MCPError: raise except Exception as e: logger.error(f"Unexpected error calling {server.name}: {e}") raise MCPError(f"Unexpected error: {str(e)}") async def list_tools(self, server_name: str) -> List[Dict[str, Any]]: """ List available tools from an MCP server. Args: server_name: Name of the MCP server Returns: List of tool definitions with name, description, and inputSchema Example: tools = await client.list_tools("iris") for tool in tools: print(f"Tool: {tool['name']} - {tool['description']}") """ result = await self._jsonrpc_call(server_name, "tools/list") tools = result.get("tools", []) logger.info(f"Listed {len(tools)} tools from {server_name}") return tools async def call_tool( self, server_name: str, tool_name: str, arguments: Dict[str, Any] ) -> Any: """ Execute a tool on an MCP server. Args: server_name: Name of the MCP server tool_name: Name of the tool to execute arguments: Tool arguments Returns: Tool execution result Example: result = await client.call_tool( "iris", "email_list_messages", {"user_email": "user@example.com", "max_results": 10} ) """ params = { "name": tool_name, "arguments": arguments } result = await self._jsonrpc_call(server_name, "tools/call", params) # MCP tools/call returns result in a specific format # Extract content from response if isinstance(result, dict) and "content" in result: content = result["content"] if isinstance(content, list) and len(content) > 0: # Get first content item first_content = content[0] if isinstance(first_content, dict) and "text" in first_content: # Parse JSON text if present try: return json.loads(first_content["text"]) except (json.JSONDecodeError, TypeError): return first_content["text"] return first_content return content return result async def get_all_tools(self) -> Dict[str, List[Dict[str, Any]]]: """ Get all tools from all registered MCP servers. Returns: Dictionary mapping server name to list of tools Example: all_tools = await client.get_all_tools() for server_name, tools in all_tools.items(): print(f"\nServer: {server_name}") for tool in tools: print(f" - {tool['name']}") """ all_tools = {} for server_name in self.servers: try: tools = await self.list_tools(server_name) all_tools[server_name] = tools except Exception as e: logger.error(f"Failed to list tools from {server_name}: {e}") all_tools[server_name] = [] return all_tools def get_server_info(self) -> Dict[str, Dict[str, Any]]: """ Get information about all registered MCP servers. Returns: Dictionary with server information """ return { name: { "url": server.url, "description": server.description, "has_api_key": server.api_key is not None } for name, server in self.servers.items() }