monoai.mcp

 12class McpServer:
 13    """MCP (Model Context Protocol) Server client for connecting to external tools.
 14    
 15    This class provides a unified interface for connecting to MCP servers using
 16    different transport protocols (HTTP/SSE, stdio). It handles connection
 17    management, tool discovery, and tool execution in both synchronous and
 18    asynchronous contexts.
 19    
 20    The MCP server allows AI agents to access external tools and services
 21    through a standardized protocol, enabling dynamic tool discovery and
 22    execution without hardcoded integrations.
 23        
 24    Examples
 25    --------
 26    >>> # HTTP/SSE server
 27    >>> server = McpServer(
 28    ...     name="coingecko",
 29    ...     server_type="http",
 30    ...     server_url="https://mcp.api.coingecko.com/sse"
 31    ... )
 32    >>> tools = server.get_tools()
 33    >>> result = server.call_tool("get_price", {"coin": "bitcoin"})
 34    
 35    >>> # Stdio server
 36    >>> server = McpServer(
 37    ...     name="local_tools",
 38    ...     server_type="python",
 39    ...     server_args=["python", "tools_server.py"]
 40    ... )
 41    >>> tools = server.get_tools()
 42    """
 43    
 44    def __init__(
 45        self,
 46        name: str,
 47        server_type: str,               # "http" | "node" | "python"
 48        server_args: list | None = None,
 49        server_url: str | None = None,
 50        env: dict | None = None,
 51        headers: dict | None = None,    # <-- per HTTP/SSE
 52        connect_timeout: float = 15.0,  # opzionale
 53    ):
 54        """Initialize the MCP server client.
 55        
 56        Parameters
 57        ----------
 58        name : str
 59            Unique identifier for this MCP server instance
 60        server_type : str
 61            Type of server connection ("http", "node", "python")
 62        server_args : list, optional
 63            Command line arguments for stdio-based servers
 64        server_url : str, optional
 65            URL for HTTP/SSE based servers
 66        env : dict, optional
 67            Environment variables for stdio-based servers
 68        headers : dict, optional
 69            HTTP headers for HTTP/SSE based servers
 70        connect_timeout : float, optional
 71            Connection timeout in seconds (default: 15.0)
 72        """
 73        self.name = name
 74        self._server_type = server_type
 75        self._server_args = server_args or []
 76        self._server_url = server_url
 77        self._env = env or {}
 78        self._headers = headers or {}
 79
 80    def _get_client(self):
 81        """Get the appropriate MCP client for the server type.
 82        
 83        Returns
 84        -------
 85        async generator
 86            MCP client context manager for the specified server type
 87            
 88        Raises
 89        ------
 90        ValueError
 91            If server_type is "http" but server_url is not provided
 92        """
 93        if self._server_type == "http":
 94            if not self._server_url:
 95                raise ValueError("URL mancante per server_type='http'")
 96            
 97            if self._server_url.endswith("/mcp"):
 98                return streamablehttp_client(self._server_url, headers=self._headers)
 99            else:
100                return sse_client(self._server_url, headers=self._headers)
101        else:
102            server_params = StdioServerParameters(
103                command="npx" if self._server_type == "node" else "python",
104                args=self._server_args,
105                env=self._env,
106                stderr="pipe",
107            )
108            
109            return stdio_client(server_params)
110
111    async def get_tools_async(self):
112        """Asynchronously retrieve available tools from the MCP server.
113        
114        Returns
115        -------
116        list
117            List of available tools with their schemas and descriptions
118            
119        Raises
120        ------
121        RuntimeError
122            If unable to connect to the server or retrieve tools
123        """
124        async with self._get_client() as (read, write):
125            async with ClientSession(read, write) as session:
126                resp = await session.list_tools()
127                return resp.tools
128
129    def get_tools(self):
130        """Synchronously retrieve available tools from the MCP server.
131        
132        This is a synchronous wrapper around the async method that runs
133        the operation in a new event loop.
134        
135        Returns
136        -------
137        list
138            List of available tools with their schemas and descriptions
139            
140        Raises
141        ------
142        RuntimeError
143            If unable to connect to the server or retrieve tools
144        """
145        return asyncio.run(self.get_tools_async())
146
147    async def call_tool_async(self, name: str, args: dict):
148        """Asynchronously execute a tool on the MCP server.
149        
150        Parameters
151        ----------
152        name : str
153            Name of the tool to execute. Can include MCP prefix (mcp_servername_toolname)
154            or just the tool name
155        args : dict
156            Arguments to pass to the tool
157            
158        Returns
159        -------
160        list
161            Tool execution result, formatted as a list of content items
162            
163        Raises
164        ------
165        RuntimeError
166            If unable to connect to the server or execute the tool
167        ValueError
168            If the tool name format is invalid
169        """
170        async with self._get_client() as (read, write):
171            async with ClientSession(read, write) as session:
172                prefix = f"mcp_{self.name}_"
173                tool_name = name[len(prefix):] if name.startswith(prefix) else name
174                result = await session.call_tool(tool_name, args)
175                                    
176                # Handle different response formats
177                if hasattr(result, 'content'):
178                    if isinstance(result.content, list) and len(result.content) > 0:
179                        if hasattr(result.content[0], 'text'):
180                            content = json.loads(result.content[0].text)
181                        else:
182                            content = str(result.content[0])
183                    else:
184                        content = str(result.content)
185                else:
186                    content = str(result)
187
188                if isinstance(content, dict):
189                    content = [content]
190                return content
191
192    def call_tool(self, name: str, args: dict):
193        """Synchronously execute a tool on the MCP server.
194        
195        This is a synchronous wrapper around the async method that runs
196        the operation in a new event loop.
197        
198        Parameters
199        ----------
200        name : str
201            Name of the tool to execute. Can include MCP prefix (mcp_servername_toolname)
202            or just the tool name
203        args : dict
204            Arguments to pass to the tool
205            
206        Returns
207        -------
208        list
209            Tool execution result, formatted as a list of content items
210            
211        Raises
212        ------
213        RuntimeError
214            If unable to connect to the server or execute the tool
215        ValueError
216            If the tool name format is invalid
217        """
218        return asyncio.run(self.call_tool_async(name, args))