Overview The Fast Agent MCP client APIs provide powerful tools for connecting to MCP (Model Context Protocol) servers, managing persistent connections, and handling server lifecycle. These APIs enable you to integrate external tools, resources, and prompts into your agent applications.
Client Session MCPAgentClientSession Extended MCP client session with enhanced features for agent frameworks.
from fast_agent.mcp import MCPAgentClientSession from fast_agent.mcp.gen_client import connect # Connect to an MCP server session = await connect( server_name = "filesystem" , server_registry = registry, client_session_factory = MCPAgentClientSession )
Name of the MCP server to connect to (must be configured in registry)
server_registry
ServerRegistryProtocol
required
Server registry containing server configurations
client_session_factory
Callable
default: "MCPAgentClientSession"
Factory function to create client session instances
Initialized MCP client session ready for communication
Session Methods Call a tool provided by the MCP server.
result = await session.call_tool( name = "read_file" , arguments = { "path" : "/path/to/file.txt" }, progress_callback = lambda progress : print (progress) )
Arguments to pass to the tool
Timeout for reading the tool response
Callback function for progress updates
Optional metadata to include in the request
Tool execution result containing output content
read_resource() Read a resource from the MCP server.
result = await session.read_resource( uri = "file:///path/to/document.txt" ) text = result.contents[ 0 ].text
URI of the resource to read
Optional metadata for the request
Resource contents (text or binary data)
get_prompt() Retrieve a prompt from the MCP server.
result = await session.get_prompt( name = "code_review" , arguments = { "language" : "python" } ) messages = result.messages
Name of the prompt to retrieve
Arguments for prompt templating
Optional metadata for the request
Prompt with messages and optional description
ping() Check server liveness with a ping request.
await session.ping( read_timeout_seconds = timedelta( seconds = 5 ))
Timeout for the ping response
Empty result indicating successful ping
Connection Management connect() Create a persistent connection to an MCP server.
from fast_agent.mcp.gen_client import connect session = await connect( server_name = "github" , server_registry = registry )
Name of the server to connect to
server_registry
ServerRegistryProtocol
required
Server registry instance
client_session_factory
Callable
default: "MCPAgentClientSession"
Factory for creating session instances
Persistent client session
disconnect() Disconnect from an MCP server or all servers.
from fast_agent.mcp.gen_client import disconnect # Disconnect from specific server await disconnect( server_name = "github" , server_registry = registry) # Disconnect from all servers await disconnect( server_name = None , server_registry = registry)
Name of server to disconnect from, or None to disconnect all
server_registry
ServerRegistryProtocol
required
Server registry instance
gen_client() Context manager for temporary server connections.
from fast_agent.mcp.gen_client import gen_client async with gen_client( "filesystem" , registry) as session: result = await session.call_tool( "read_file" , { "path" : "data.txt" }) # Session automatically closed on exit
Name of the server to connect to
server_registry
ServerRegistryProtocol
required
Server registry instance
client_session_factory
Callable
default: "MCPAgentClientSession"
Factory for creating session instances
Temporary client session (auto-closed on exit)
Connection Manager MCPConnectionManager Manages the lifecycle of multiple MCP server connections.
from fast_agent.mcp.mcp_connection_manager import MCPConnectionManager manager = MCPConnectionManager( server_registry = registry) async with manager: # Get or create server connection conn = await manager.get_server( server_name = "github" , client_session_factory = MCPAgentClientSession, startup_timeout_seconds = 30.0 ) session = conn.session
get_server() Get a running server instance, launching it if needed.
Name of the server to get or launch
Factory function for creating client sessions
Maximum time to wait for server startup (excluding OAuth)
Whether to trigger OAuth flow if configured
Custom handler for OAuth events
allow_oauth_paste_fallback
Allow manual token paste if OAuth flow fails
Server connection object with session and metadata
disconnect_server() Disconnect a specific server.
await manager.disconnect_server( "github" )
Name of the server to disconnect
reconnect_server() Force reconnection to a server (useful after errors).
conn = await manager.reconnect_server( server_name = "github" , client_session_factory = MCPAgentClientSession )
Name of the server to reconnect
Factory function for creating client sessions
Maximum time to wait for reconnection
New server connection instance
disconnect_all() Disconnect all running servers.
await manager.disconnect_all()
Server Connection ServerConnection Represents a long-lived MCP server connection.
conn = await manager.get_server( "filesystem" , MCPAgentClientSession) # Access session session = conn.session # Check health if conn.is_healthy(): result = await session.call_tool( ... ) # Access server metadata print (conn.server_instructions) print (conn.server_capabilities) print (conn.session_id)
Active MCP client session
Name of the connected server
Server configuration settings
server_capabilities
ServerCapabilities | None
Capabilities advertised by the server
Instructions provided by the server (if enabled)
Server implementation information
Current session identifier
is_healthy() Check if the connection is healthy and ready to use.
if conn.is_healthy(): # Safe to use the session pass
True if connection is active and error-free
Error Handling ServerInitializationError Raised when server initialization fails.
from fast_agent.core.exceptions import ServerInitializationError try : conn = await manager.get_server( "myserver" , MCPAgentClientSession) except ServerInitializationError as e: print ( f "Failed to initialize: { e } " ) print ( f "Details: { e.details } " )
ServerSessionTerminatedError Raised when a server session is terminated (e.g., 404 from server restart).
from fast_agent.core.exceptions import ServerSessionTerminatedError try : result = await session.call_tool( "my_tool" , {}) except ServerSessionTerminatedError: # Reconnect to server conn = await manager.reconnect_server( "myserver" , MCPAgentClientSession) session = conn.session
Best Practices Use persistent connections with connect() for long-running applications. For one-off operations, use the gen_client() context manager to ensure proper cleanup.
Always check is_healthy() before using a server connection, especially after catching exceptions. Use reconnect_server() to recover from connection errors.
The connection manager automatically handles OAuth flows, server lifecycle, and ping monitoring. Configure these settings in your server registry.
Example: Complete MCP Integration from fast_agent import ServerRegistry from fast_agent.mcp.gen_client import connect, disconnect from fast_agent.mcp import MCPAgentClientSession # Initialize registry registry = ServerRegistry() registry.add_server( name = "github" , url = "https://github-mcp.example.com" , transport = "sse" ) # Connect to server session = await connect( "github" , registry) try : # List available tools tools_result = await session.list_tools() # Call a tool result = await session.call_tool( name = "create_issue" , arguments = { "repo" : "owner/repo" , "title" : "Bug report" , "body" : "Description of the issue" } ) # Read a resource resource = await session.read_resource( uri = "github://owner/repo/README.md" ) finally : # Clean up await disconnect( "github" , registry)
Prompts Load and manage MCP prompts
Resources Work with MCP resources
