connect_to_server
Establish a connection to a target MCP server for testing purposes using stdio or HTTP transport protocols. Verifies connectivity and returns detailed connection state information.
Instructions
Connect to an MCP server for testing.
Establishes a connection to a target MCP server using the appropriate transport protocol (stdio for file paths, streamable-http for URLs). Only one connection can be active at a time.
Returns: Dictionary with connection details including: - success: Always True on successful connection - connection: Full ConnectionState with server info and statistics - message: Human-readable success message - metadata: Request timing information
Raises: Returns error dict on failure with: - success: False - error: Error details (type, message, suggestion) - metadata: Request timing information
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| url | Yes | Server URL (http://..., https://...) or file path for stdio transport |
Implementation Reference
- The primary handler for the 'connect_to_server' MCP tool. Decorated with @mcp.tool(), it accepts a server URL, uses ConnectionManager to connect, provides contextual updates via ctx.info/error, handles exceptions with detailed error responses, and returns structured JSON-compatible results including connection state and metadata.
@mcp.tool async def connect_to_server( url: Annotated[str, "Server URL (http://..., https://...) or file path for stdio transport"], ctx: Context ) -> dict[str, Any]: """Connect to an MCP server for testing. Establishes a connection to a target MCP server using the appropriate transport protocol (stdio for file paths, streamable-http for URLs). Only one connection can be active at a time. Returns: Dictionary with connection details including: - success: Always True on successful connection - connection: Full ConnectionState with server info and statistics - message: Human-readable success message - metadata: Request timing information Raises: Returns error dict on failure with: - success: False - error: Error details (type, message, suggestion) - metadata: Request timing information """ start_time = time.perf_counter() try: # User-facing progress update await ctx.info(f"Connecting to MCP server at {url}") # Detailed technical log logger.info(f"Connecting to MCP server at: {url}") state: ConnectionState = await ConnectionManager.connect(url) elapsed_ms = (time.perf_counter() - start_time) * 1000 # User-facing success update await ctx.info(f"Successfully connected to {url}") # Detailed technical log logger.info( f"Successfully connected to {url}", extra={ "url": url, "transport": state.transport, "duration_ms": elapsed_ms, }, ) return { "success": True, "connection": state.model_dump(mode="json"), "message": f"Successfully connected to {url}", "metadata": { "request_time_ms": round(elapsed_ms, 2), "transport": state.transport, "server_url": state.server_url, }, } except ConnectionError as e: elapsed_ms = (time.perf_counter() - start_time) * 1000 # User-facing error update await ctx.error(f"Failed to connect to {url}: {str(e)}") # Detailed technical log logger.error( f"Failed to connect to {url}: {str(e)}", extra={"url": url, "error": str(e), "duration_ms": elapsed_ms}, ) # Determine appropriate suggestion based on error suggestion = "Check that the server URL is correct and the server is running" if "timed out" in str(e).lower(): suggestion = "The connection timed out. Check server availability and network connectivity" elif "file" in url.lower() or not url.startswith("http"): suggestion = "For file paths, ensure the path is valid and the server executable has correct permissions" return { "success": False, "error": { "error_type": "connection_failed", "message": str(e), "details": {"url": url}, "suggestion": suggestion, }, "connection": None, "metadata": { "request_time_ms": round(elapsed_ms, 2), "attempted_url": url, }, } except Exception as e: elapsed_ms = (time.perf_counter() - start_time) * 1000 # User-facing error update await ctx.error(f"Unexpected error connecting to {url}: {str(e)}") # Detailed technical log logger.exception( f"Unexpected error connecting to {url}", extra={"url": url, "duration_ms": elapsed_ms}, ) return { "success": False, "error": { "error_type": "connection_failed", "message": f"Unexpected error: {str(e)}", "details": {"url": url, "exception_type": type(e).__name__}, "suggestion": "This is an unexpected error. Check server logs for more details", }, "connection": None, "metadata": { "request_time_ms": round(elapsed_ms, 2), "attempted_url": url, }, } - ConnectionManager.connect() - Core helper method that implements the MCP client connection logic. Creates FastMCP Client, establishes async connection with timeout, infers transport type, retrieves server info/capabilities, initializes and stores ConnectionState singleton, and manages prior connections by disconnecting them.
async def connect(cls, url: str) -> ConnectionState: """Connect to an MCP server. Creates a FastMCP Client instance, establishes connection, and stores the connection state globally. If a connection already exists, it will be closed before establishing the new connection. Args: url: Server URL or file path - HTTP/HTTPS URLs use streamable-http transport - File paths use stdio transport Returns: ConnectionState with connection details and initial statistics Raises: ConnectionError: If connection fails """ async with _connection.lock: # Close existing connection if any if _connection.client is not None: await cls._disconnect_internal() # Get timeout configuration connect_timeout = cls._get_timeout("MCP_TEST_CONNECT_TIMEOUT", 30.0) try: # Create client - let FastMCP auto-detect transport client = Client(url, timeout=connect_timeout) # Establish connection await asyncio.wait_for(client.__aenter__(), timeout=connect_timeout) # Infer transport type transport = cls._infer_transport(url) # Get server information server_info: dict[str, Any] = {} try: # Try to get server info via initialization result if hasattr(client, "_session") and client._session: session = client._session if hasattr(session, "server_info"): info = session.server_info server_info = { "name": getattr(info, "name", None), "version": getattr(info, "version", None), } # Add capabilities if available if hasattr(session, "server_capabilities"): caps: ServerCapabilities = session.server_capabilities server_info["capabilities"] = { "tools": bool(getattr(caps, "tools", None)), "resources": bool(getattr(caps, "resources", None)), "prompts": bool(getattr(caps, "prompts", None)), } except Exception: # If we can't get server info, continue without it pass # Create connection state state = ConnectionState( server_url=url, transport=transport, # type: ignore connected_at=datetime.now(), server_info=server_info if server_info else None, statistics={ "tools_called": 0, "resources_accessed": 0, "prompts_executed": 0, "errors": 0, }, ) # Store globally _connection.client = client _connection.state = state return state except asyncio.TimeoutError as e: raise ConnectionError( f"Connection to {url} timed out after {connect_timeout}s" ) from e except Exception as e: raise ConnectionError(f"Failed to connect to {url}: {str(e)}") from e - Creation of the shared FastMCP(name="mcp-test-mcp") server instance named 'mcp'. All @mcp.tool() decorators in tool modules register tools against this instance.
# Create the shared FastMCP server instance mcp = FastMCP(name="mcp-test-mcp") - node-wrapper/python-src/src/mcp_test_mcp/server.py:96-99 (registration)Import of tool modules (including 'connection') after mcp instance, which executes the module-level @mcp.tool() decorators to automatically register the 'connect_to_server' tool on the FastMCP server.
# Import tool modules to trigger decorator registration # This MUST happen after mcp instance is imported but before the server runs from .tools import connection, tools, resources, prompts, llm - ConnectionState Pydantic model defining the structured connection data returned by connect_to_server, including server_url, transport, connected_at, server_info, and statistics. Used for input/output validation and serialization.
class ConnectionState(BaseModel): """Current state of an MCP server connection. This model captures all relevant information about an active or attempted connection to an MCP server. """ server_url: str = Field(description="The URL or identifier of the MCP server") transport: Literal["stdio", "sse", "streamable-http"] = Field( description="The transport protocol used for communication" ) connected_at: Optional[datetime] = Field( default=None, description="Timestamp when connection was established, None if not connected", ) server_info: Optional[dict[str, Any]] = Field( default=None, description="Server metadata returned during initialization (name, version, capabilities)", ) statistics: dict[str, int] = Field( default_factory=lambda: { "tools_called": 0, "resources_accessed": 0, "prompts_executed": 0, "errors": 0, }, description="Usage statistics for this connection", )