MCP Client Development with Python SDK

MCP Client Development - with Python SDK

Introduction and Setup

This tutorial guides you through developing MCP (Model Context Protocol) clients using the Python SDK. MCP clients connect to MCP servers to leverage their Resources, Prompts, and Tools, forming the backbone of context-aware AI applications.

The Python SDK provides a Pythonic interface, leverages asyncio for non-blocking operations, and is ideal for building backend services, scripts, or any Python application needing to interact with MCP servers.

Prerequisites:

• Python (version 3.8 or later recommended).
• pip (Python’s package installer).

Installation:

Install the MCP Python SDK package from PyPI:

pip install modelcontext-mcp

Connecting to an MCP Server

The MCPClient class is used to manage connections. It’s best practice to use an async with statement, which ensures the connection is properly established and closed.

import asyncio
from modelcontext_mcp import MCPClient, MCPError

async def connect_and_operate():
    server_uri = 'ws://localhost:8080/mcp'  # Replace with your MCP server's WebSocket URI
    try:
        async with MCPClient(
            uri=server_uri,
            client_id='my-python-client-01',
            client_description='An example MCP client using the Python SDK'
            # You can also declare client capabilities here
        ) as client:
            print(f'Successfully connected to MCP server at {server_uri}!')
            print(f'Server capabilities: {client.server_capabilities}')

            # Client is now connected and ready for operations.
            # We'll add operations in the next sections.
            pass # Placeholder for operations

    except MCPError as e:
        print(f'MCP Client Error (Code: {e.code}): {e.message}. Data: {e.data}')
    except ConnectionRefusedError:
        print(f'Connection refused. Is the MCP server running at {server_uri}?')
    except Exception as e:
        print(f'An unexpected error occurred: {e}')
    finally:
        print('Client operations finished or an error occurred.')

# To run this (and subsequent examples):
# if __name__ == "__main__":
#     asyncio.run(connect_and_operate())

The async with block handles client.connect() on entry and client.close() on exit automatically.

Interacting with Server Capabilities

With an active client connection, you can interact with the server’s Resources, Prompts, and Tools using the client’s request_* and stream_* methods.

Requesting Resources: To fetch data from a resource, use client.request_resource(resource_id, params) for a single response, or iterate over client.stream_resource(resource_id, params) for streaming data.

# Inside an async function, assuming 'client' is an active MCPClient instance

async def get_server_file(client: MCPClient):
    try:
        # Assuming server has a 'file_content' resource
        # that takes {'path': 'file_path.txt'} and returns {'content': '...'}
        response = await client.request_resource(
            resource_id='file_content',
            params={'path': '/data/info.txt'}
        )
        print(f"Resource (file_content) response: {response.payload.get('content')}")
    except MCPError as e:
        print(f'Error requesting resource: {e.message}')

async def stream_server_logs(client: MCPClient):
    try:
        # Assuming server has 'live_logs' resource that streams {'line': 'log line'}
        print('Streaming server logs...')
        async for update in client.stream_resource(
            resource_id='live_logs',
            params={'level': 'INFO'}
        ):
            print(f"Log: {update.payload.get('line')}")
            if "END_STREAM" in update.payload.get('line', ''):
                 print("End of stream marker found.")
                 break
        print('Log streaming finished.')
    except MCPError as e:
        print(f'Error streaming resource: {e.message}')

Utilizing Prompts: To send data to a prompt (e.g., for LLM processing), use client.request_prompt(prompt_id, params) for a single output or client.stream_prompt(prompt_id, params) for streaming outputs like token-by-token generation.

# Inside an async function, assuming 'client' is an active MCPClient instance

async def get_text_completion(client: MCPClient, user_input: str):
    try:
        # Assuming 'text_generator' prompt takes {'input_text': '...'}
        # and returns {'completion': '...'}
        response = await client.request_prompt(
            prompt_id='text_generator',
            params={'input_text': user_input, 'max_tokens': 50}
        )
        print(f"Prompt (text_generator) completion: {response.payload.get('completion')}")
    except MCPError as e:
        print(f'Error requesting prompt: {e.message}')

async def stream_generated_text(client: MCPClient, topic: str):
    try:
        # Assuming 'creative_writer' streams {'token': '...'}
        print(f'Streaming creative writing on "{topic}"...')
        full_text = []
        async for update in client.stream_prompt(
            prompt_id='creative_writer',
            params={'topic': topic}
        ):
            token = update.payload.get('token', '')
            print(token, end='', flush=True)
            full_text.append(token)
        print('\nCreative writing stream finished.')
        # print(f"Full text: {''.join(full_text)}")
    except MCPError as e:
        print(f'Error streaming prompt: {e.message}')

Invoking Tools: To execute a server-side function, use client.request_tool(tool_id, params).

# Inside an async function, assuming 'client' is an active MCPClient instance

async def execute_remote_calculation(client: MCPClient, num1: int, num2: int):
    try:
        # Assuming 'math_tool' takes {'a': N, 'b': N, 'op': 'name'}
        # and returns {'result': M}
        response = await client.request_tool(
            tool_id='math_tool',
            params={'a': num1, 'b': num2, 'op': 'multiply'}
        )
        print(f"Tool (math_tool) result: {response.payload.get('result')}")
    except MCPError as e:
        print(f'Error invoking tool: {e.message}')

Error Handling

Proper error handling is vital. Catch MCPError for protocol-specific issues (like invalid parameters or server-side execution errors) and standard Python exceptions (like ConnectionRefusedError) for network or other problems.

# General structure for error handling in an operation
# client: MCPClient
try:
    # response = await client.request_resource(...)
    pass # Your MCP operation here
except MCPError as e:
    # Specific MCP error from the server or protocol violation
    print(f'MCP Error! Code: {e.code}, Message: {e.message}, Details: {e.data}')
except ConnectionRefusedError:
    print('Connection was refused by the server. Is it running and accessible?')
except asyncio.TimeoutError:
    print('The operation timed out.')
except Exception as e:
    # Other unexpected errors
    print(f'An unexpected Python error occurred: {type(e).__name__} - {e}')

Full Client Example

This example combines the concepts: connects to a server, requests a resource, utilizes a prompt, invokes a tool, and handles potential errors. It assumes an MCP server is running at ws://localhost:8080/mcp and exposes the specified capabilities.

import asyncio
from modelcontext_mcp import MCPClient, MCPError

async def run_full_python_client():
    server_uri = 'ws://localhost:8080/mcp'
    try:
        async with MCPClient(
            uri=server_uri,
            client_id='py-full-client-example',
            client_description='Full Python MCP Client Example'
        ) as client:
            print(f'Connected to {server_uri}. Server Info: {client.server_capabilities}')

            # 1. Request a 'system_status' resource
            # Assumes payload: {'status': '...', 'load': 0.0}
            print("\nRequesting 'system_status' resource...")
            status_payload = await client.request_resource(resource_id='system_status', params={})
            print(f"System Status: {status_payload.payload.get('status')}, Load: {status_payload.payload.get('load')}")

            # 2. Utilize a 'translate' prompt
            # Assumes params: {'text': '...', 'target_lang': '...'}, payload: {'translation': '...'}
            text_to_translate = "Hello, MCP world!"
            print(f"\nRequesting 'translate' prompt for: '{text_to_translate}' to Spanish...")
            translate_payload = await client.request_prompt(
                prompt_id='translate',
                params={'text': text_to_translate, 'target_lang': 'es'}
            )
            print(f"Translation: {translate_payload.payload.get('translation')}")

            # 3. Invoke a 'database_query' tool
            # Assumes params: {'query': 'SELECT ...'}, payload: {'rows': [...]}
            print("\nInvoking 'database_query' tool...")
            query_result = await client.request_tool(
                tool_id='database_query',
                params={'query': 'SELECT COUNT(*) FROM users;'}
            )
            print(f"Query Result (User Count): {query_result.payload.get('rows')}")

    except MCPError as e:
        print(f'MCP Communication Error! Code: {e.code}, Message: {e.message}, Data: {e.data}')
    except ConnectionRefusedError:
        print(f'Connection refused at {server_uri}. Please ensure the MCP server is running.')
    except Exception as e:
        print(f'An unexpected error occurred: {type(e).__name__} - {e}')
    finally:
        print('\nPython client operations concluded.')

if __name__ == "__main__":
    asyncio.run(run_full_python_client())

Best Practices for Python MCP Clients

• Context Management: Always use async with MCPClient(...) as client: to ensure connections are properly established and torn down, even if errors occur.
• Comprehensive Error Handling: Catch MCPError specifically for protocol-level issues and be prepared for common network exceptions like ConnectionRefusedError or asyncio.TimeoutError.
• Asynchronous Code: Write all client code that interacts with the MCP server using async and await to leverage Python’s asyncio library for efficient, non-blocking I/O.
• Payload Validation: While Python is dynamically typed, consider validating the structure or presence of expected keys in response.payload dictionary, especially when interacting with unfamiliar servers.
• Streaming for Efficiency: Use client.stream_resource and client.stream_prompt when dealing with potentially large datasets or real-time data feeds to avoid high memory usage and improve responsiveness.
• Understand Server Capabilities: Inspect client.server_capabilities after connection to programmatically understand what the connected server offers. This can make your client more adaptive.