(self,
server_name: str,
timeout: int = CONNECTION_TIMEOUT,
**kwargs)
| 141 | f'with tools: \n{sep.join(tools)}.') |
| 142 | |
| 143 | async def connect_to_server(self, |
| 144 | server_name: str, |
| 145 | timeout: int = CONNECTION_TIMEOUT, |
| 146 | **kwargs): |
| 147 | logger.info(f'connect to {server_name}') |
| 148 | # transport: stdio, sse, streamable_http, websocket |
| 149 | transport = kwargs.get('transport') or kwargs.get('type') |
| 150 | command = kwargs.get('command') |
| 151 | url = kwargs.get('url') |
| 152 | session_kwargs = kwargs.get('session_kwargs') |
| 153 | if url: |
| 154 | if transport and transport.lower() == 'sse': |
| 155 | logger.info( |
| 156 | '`transport` or `type` is configured as "sse", using sse transport.' |
| 157 | ) |
| 158 | sse_transport = await self.exit_stack.enter_async_context( |
| 159 | sse_client( |
| 160 | url, kwargs.get('headers'), |
| 161 | kwargs.get('timeout', DEFAULT_HTTP_TIMEOUT), |
| 162 | kwargs.get('sse_read_timeout', |
| 163 | DEFAULT_SSE_READ_TIMEOUT))) |
| 164 | read, write = sse_transport |
| 165 | |
| 166 | elif transport and transport.lower() == 'websocket': |
| 167 | logger.info( |
| 168 | '`transport` or `type` is configured as "websocket", using websocket transport.' |
| 169 | ) |
| 170 | try: |
| 171 | from mcp.client.websocket import websocket_client |
| 172 | except ImportError: |
| 173 | raise ImportError( |
| 174 | 'Could not import websocket_client. ' |
| 175 | 'To use Websocket connections, please install the required dependency with: ' |
| 176 | "'pip install mcp[ws]' or 'pip install websockets'" |
| 177 | ) from None |
| 178 | websocket_transport = await self.exit_stack.enter_async_context( |
| 179 | websocket_client(url)) |
| 180 | read, write = websocket_transport |
| 181 | |
| 182 | else: |
| 183 | logger.info( |
| 184 | 'Using streamable_http transport. To configure a different transport such as sse, please' |
| 185 | 'set the `type` or `transport` variable to "sse".') |
| 186 | try: |
| 187 | from mcp.client.streamable_http import streamablehttp_client |
| 188 | except ImportError: |
| 189 | raise ImportError( |
| 190 | 'Could not import streamablehttp_client. ' |
| 191 | 'To use streamable http connections, please upgrade to the latest version of mcp with: ' |
| 192 | "'pip install -U mcp'") from None |
| 193 | httpx_client_factory = kwargs.get('httpx_client_factory') |
| 194 | other_kwargs = {} |
| 195 | if httpx_client_factory is not None: |
| 196 | other_kwargs['httpx_client_factory'] = httpx_client_factory |
| 197 | streamable_transport = await self.exit_stack.enter_async_context( |
| 198 | streamablehttp_client( |
| 199 | url, |
| 200 | headers=kwargs.get('headers'), |
no test coverage detected