tessl/pypi-mcp
Model Context Protocol SDK for building MCP servers and clients in Python
—
Pending
client.mddocs/
Client Operations
Core client functionality for connecting to and interacting with MCP servers. The client APIs provide session management, tool calling, resource access, and prompt handling with support for multiple concurrent server connections.
Capabilities
ClientSession
Primary client class for managing individual connections to MCP servers with full protocol support.
class ClientSession:
def __init__(
self,
read_stream: MemoryObjectReceiveStream,
write_stream: MemoryObjectSendStream,
read_timeout_seconds: timedelta | None = None,
sampling_callback: Callable | None = None,
elicitation_callback: Callable | None = None,
list_roots_callback: Callable | None = None,
logging_callback: Callable | None = None,
message_handler: Callable | None = None,
client_info: Implementation | None = None,
):
"""
Initialize a client session for communicating with an MCP server.
Parameters:
- read_stream: Stream for receiving messages from server
- write_stream: Stream for sending messages to server
- read_timeout_seconds: Timeout for read operations
- sampling_callback: Callback for LLM sampling requests
- elicitation_callback: Callback for user input requests
- list_roots_callback: Callback for root directory listing
- logging_callback: Callback for server log messages
- message_handler: Custom message handler
- client_info: Client implementation metadata
"""
async def initialize(self) -> InitializeResult:
"""
Initialize the MCP connection with the server.
Returns:
InitializeResult containing server capabilities and metadata
"""
async def list_tools(self, cursor: str | None = None) -> ListToolsResult:
"""
List all available tools from the server.
Parameters:
- cursor: Pagination cursor for continuation
Returns:
ListToolsResult containing tools and optional next cursor
"""
async def call_tool(
self,
name: str,
arguments: dict[str, Any] | None = None
) -> CallToolResult:
"""
Call a specific tool with provided arguments.
Parameters:
- name: Name of the tool to call
- arguments: Tool arguments as key-value pairs
Returns:
CallToolResult containing tool output content
"""
async def list_resources(self, cursor: str | None = None) -> ListResourcesResult:
"""
List all available resources from the server.
Parameters:
- cursor: Pagination cursor for continuation
Returns:
ListResourcesResult containing resources and optional next cursor
"""
async def list_resource_templates(self, cursor: str | None = None) -> ListResourceTemplatesResult:
"""
List all available resource templates from the server.
Parameters:
- cursor: Pagination cursor for continuation
Returns:
ListResourceTemplatesResult containing resource templates and optional next cursor
"""
async def read_resource(self, uri: AnyUrl) -> ReadResourceResult:
"""
Read the content of a specific resource.
Parameters:
- uri: URI of the resource to read
Returns:
ReadResourceResult containing resource content
"""
async def subscribe_resource(self, uri: AnyUrl) -> EmptyResult:
"""
Subscribe to updates for a specific resource.
Parameters:
- uri: URI of the resource to subscribe to
Returns:
EmptyResult indicating success
"""
async def unsubscribe_resource(self, uri: AnyUrl) -> EmptyResult:
"""
Unsubscribe from updates for a specific resource.
Parameters:
- uri: URI of the resource to unsubscribe from
Returns:
EmptyResult indicating success
"""
async def list_prompts(self, cursor: str | None = None) -> ListPromptsResult:
"""
List all available prompts from the server.
Parameters:
- cursor: Pagination cursor for continuation
Returns:
ListPromptsResult containing prompts and optional next cursor
"""
async def get_prompt(
self,
name: str,
arguments: dict[str, str] | None = None
) -> GetPromptResult:
"""
Get a specific prompt with provided arguments.
Parameters:
- name: Name of the prompt to get
- arguments: Prompt arguments as key-value pairs
Returns:
GetPromptResult containing prompt messages
"""
async def set_logging_level(self, level: LoggingLevel) -> EmptyResult:
"""
Set the logging level for server messages.
Parameters:
- level: Logging level (DEBUG, INFO, NOTICE, WARNING, ERROR, CRITICAL, ALERT, EMERGENCY)
Returns:
EmptyResult indicating success
"""
async def send_progress_notification(
self,
progress_token: ProgressToken,
progress: float,
total: float | None = None
) -> None:
"""
Send progress notification to server.
Parameters:
- progress_token: Token identifying the operation
- progress: Current progress value
- total: Total expected value (optional)
"""
async def send_ping(self) -> EmptyResult:
"""
Send ping request to server for connection health check.
Returns:
EmptyResult indicating server responded to ping
"""
async def complete(
self,
ref: CompletionReference,
argument: CompletionArgument | None = None
) -> CompleteResult:
"""
Request completion suggestions for prompts or resources.
Parameters:
- ref: Reference to prompt or resource for completion
- argument: Optional completion argument context
Returns:
CompleteResult containing completion suggestions
"""
async def send_roots_list_changed(self) -> None:
"""
Send notification that the roots directory listing has changed.
Used to inform the server about changes in available root directories.
"""ClientSessionGroup
Multi-session manager for coordinating connections to multiple MCP servers simultaneously with aggregated tool/resource access.
class ClientSessionGroup:
def __init__(
self,
exit_stack: AsyncExitStack | None = None,
component_name_hook: Callable | None = None,
):
"""
Initialize a client session group for managing multiple server connections.
Parameters:
- exit_stack: Async context manager stack for cleanup
- component_name_hook: Hook for customizing component names
"""
async def connect_to_server(self, server_params: ServerParameters) -> ClientSession:
"""
Connect to an MCP server and add to the group.
Parameters:
- server_params: Server connection parameters
Returns:
ClientSession for the connected server
"""
async def disconnect_from_server(self, session: ClientSession) -> None:
"""
Disconnect from a server and remove from the group.
Parameters:
- session: ClientSession to disconnect
"""
async def call_tool(self, name: str, args: dict[str, Any]) -> CallToolResult:
"""
Call a tool from any connected server by name.
Parameters:
- name: Name of the tool to call
- args: Tool arguments
Returns:
CallToolResult from the server that provides the tool
"""
@property
def sessions(self) -> list[ClientSession]:
"""List of all active client sessions."""
@property
def tools(self) -> dict[str, Tool]:
"""Dictionary of all available tools from all connected servers."""
@property
def resources(self) -> dict[str, Resource]:
"""Dictionary of all available resources from all connected servers."""
@property
def prompts(self) -> dict[str, Prompt]:
"""Dictionary of all available prompts from all connected servers."""Transport Client Functions
Functions for creating client transport connections to MCP servers.
def stdio_client(server: StdioServerParameters) -> AsyncContextManager:
"""
Create a stdio transport client connection.
Parameters:
- server: Stdio server parameters
Returns:
Async context manager yielding (read_stream, write_stream)
"""
def websocket_client(url: str, **kwargs) -> AsyncContextManager:
"""
Create a WebSocket client connection.
Parameters:
- url: WebSocket server URL
- **kwargs: Additional connection options
Returns:
Async context manager yielding (read_stream, write_stream)
"""
def sse_client(
url: str,
headers: dict[str, str] | None = None,
**kwargs
) -> AsyncContextManager:
"""
Create a Server-Sent Events client connection.
Parameters:
- url: SSE endpoint URL
- headers: HTTP headers for connection
- **kwargs: Additional connection options
Returns:
Async context manager yielding (read_stream, write_stream)
"""
def streamablehttp_client(
url: str,
headers: dict[str, str] | None = None,
**kwargs
) -> AsyncContextManager:
"""
Create a streaming HTTP client connection.
Parameters:
- url: HTTP endpoint URL
- headers: HTTP headers for connection
- **kwargs: Additional connection options
Returns:
Async context manager yielding (read_stream, write_stream)
"""Transport Configuration Classes
Configuration classes for different transport types.
class StdioServerParameters:
def __init__(
self,
command: str | list[str],
args: list[str] | None = None,
env: dict[str, str] | None = None,
cwd: str | None = None,
):
"""
Parameters for stdio server connection.
Parameters:
- command: Server command to execute
- args: Command arguments
- env: Environment variables
- cwd: Working directory
"""
class SseServerParameters:
def __init__(
self,
url: str,
headers: dict[str, str] | None = None,
**kwargs
):
"""
Parameters for Server-Sent Events connection.
Parameters:
- url: SSE endpoint URL
- headers: HTTP headers
- **kwargs: Additional options
"""
class StreamableHttpParameters:
def __init__(
self,
url: str,
headers: dict[str, str] | None = None,
**kwargs
):
"""
Parameters for streaming HTTP connection.
Parameters:
- url: HTTP endpoint URL
- headers: HTTP headers
- **kwargs: Additional options
"""Client Authentication
Client-side authentication utilities for connecting to secured MCP servers.
class ClientAuth:
def __init__(self, token: str | None = None, **kwargs):
"""
Client authentication handler.
Parameters:
- token: Bearer token for authentication
- **kwargs: Additional auth options
"""
async def get_headers(self) -> dict[str, str]:
"""
Get authentication headers for requests.
Returns:
Dictionary of HTTP headers for authentication
"""
class AuthenticationError(Exception):
"""Exception raised for authentication failures."""Usage Examples
Basic Client Connection
import asyncio
import mcp
async def basic_client():
# Connect to a stdio server
server_params = mcp.StdioServerParameters(
command=["python", "my_server.py"]
)
async with mcp.stdio_client(server_params) as (read, write):
async with mcp.ClientSession(read, write) as session:
# Initialize connection
result = await session.initialize()
print(f"Connected to: {result.server_info.name}")
# List and call tools
tools = await session.list_tools()
if tools.tools:
tool_result = await session.call_tool(
tools.tools[0].name,
{"input": "test"}
)
print(f"Tool result: {tool_result}")
asyncio.run(basic_client())Multi-Server Management
import asyncio
import mcp
async def multi_server_client():
async with mcp.ClientSessionGroup() as group:
# Connect to multiple servers
server1 = mcp.StdioServerParameters(command=["server1"])
server2 = mcp.StdioServerParameters(command=["server2"])
session1 = await group.connect_to_server(server1)
session2 = await group.connect_to_server(server2)
# Access aggregated tools from all servers
print(f"All tools: {list(group.tools.keys())}")
# Call tools from any server
result = await group.call_tool("any_tool_name", {"param": "value"})
print(f"Result: {result}")
asyncio.run(multi_server_client())WebSocket Client
import asyncio
import mcp
async def websocket_client_example():
async with mcp.websocket_client("ws://localhost:8000/mcp") as (read, write):
async with mcp.ClientSession(read, write) as session:
await session.initialize()
# Use WebSocket connection
resources = await session.list_resources()
for resource in resources.resources:
content = await session.read_resource(resource.uri)
print(f"Resource {resource.name}: {content}")
asyncio.run(websocket_client_example())Install with Tessl CLI
npx tessl i tessl/pypi-mcp