Represents a connection to an MCP server
| 259 | raise |
| 260 | |
| 261 | class MCPServer: |
| 262 | """Represents a connection to an MCP server""" |
| 263 | |
| 264 | def __init__(self, server_name: str, config: ServerConfig): |
| 265 | self.server_name = server_name |
| 266 | self.config = config |
| 267 | self.tools = [] |
| 268 | self.resources = [] |
| 269 | self.prompts = [] |
| 270 | self.connected = False |
| 271 | self.has_tools_capability = False |
| 272 | self.has_resources_capability = False |
| 273 | self.has_prompts_capability = False |
| 274 | |
| 275 | async def connect_stdio(self, session: LoggingClientSession) -> bool: |
| 276 | """Connect to server using stdio transport and discover capabilities""" |
| 277 | try: |
| 278 | logger.info(f"Connected to server: {self.server_name}") |
| 279 | |
| 280 | # Initialize session |
| 281 | logger.debug(f"Initializing MCP session for {self.server_name}") |
| 282 | result = await session.initialize() |
| 283 | logger.info(f"Server {self.server_name} initialized with capabilities: {result.capabilities}") |
| 284 | logger.debug(f"Full initialization result: {result}") |
| 285 | |
| 286 | # Check which capabilities the server supports |
| 287 | server_capabilities = result.capabilities |
| 288 | |
| 289 | # Discover tools if supported |
| 290 | if hasattr(server_capabilities, "tools"): |
| 291 | self.has_tools_capability = True |
| 292 | logger.info(f"Discovering tools for {self.server_name}") |
| 293 | try: |
| 294 | tools_result = await session.list_tools() |
| 295 | self.tools = tools_result.tools |
| 296 | logger.info(f"Found {len(self.tools)} tools") |
| 297 | logger.debug(f"Tools details: {[t.name for t in self.tools]}") |
| 298 | except McpError as e: |
| 299 | logger.warning(f"Failed to list tools: {e}") |
| 300 | |
| 301 | # Discover resources if supported |
| 302 | if hasattr(server_capabilities, "resources"): |
| 303 | self.has_resources_capability = True |
| 304 | logger.info(f"Discovering resources for {self.server_name}") |
| 305 | try: |
| 306 | resources_result = await session.list_resources() |
| 307 | self.resources = resources_result.resources |
| 308 | logger.info(f"Found {len(self.resources)} resources") |
| 309 | logger.debug(f"Resources details: {[r.uri for r in self.resources]}") |
| 310 | except McpError as e: |
| 311 | logger.warning(f"Failed to list resources: {e}") |
| 312 | |
| 313 | # Discover prompts if supported |
| 314 | if hasattr(server_capabilities, "prompts"): |
| 315 | self.has_prompts_capability = True |
| 316 | logger.info(f"Discovering prompts for {self.server_name}") |
| 317 | try: |
| 318 | prompts_result = await session.list_prompts() |
no outgoing calls