Mcpwire
MCP client implementation
Ask AI about Mcpwire
Powered by Claude · Grounded in docs
I know everything about Mcpwire. Ask me about installation, configuration, usage, or troubleshooting.
0/500
Reviews
Documentation
MCP Python Client
A Python client library for interacting with servers implementing the Model Context Protocol (MCP). This library simplifies communication with AI models and agents that expose an MCP interface.
Version: 0.5.1
Features
- Official MCP Implementation: This library uses the official MCP library and langchain-mcp-adapters.
- Resource Support: Complete support for MCP Resources API for exposing and accessing data.
- Send Prompts: Easily send conversation history to get responses from MCP servers.
- Tool Support: Access tools provided by MCP servers and convert them to LangChain tools.
- MultiServerMCPClient: Connect to multiple MCP servers and access their tools and resources.
- Configuration File: Load server connection details and resource configuration from an
mcp.jsonfile. - Robust Error Handling: Differentiates between connection errors, timeouts, API errors, and data validation errors.
- Authentication: Supports API Key authentication via the
Authorization: Bearerheader. - Context Manager: Supports use as an async context manager.
Installation
# From PyPI (once published)
pip install mcpwire
# Development dependencies
pip install mcpwire[dev]
# Or directly from GitHub
pip install git+https://github.com/anukchat/mcpwire.git
# Or for development
git clone https://github.com/anukchat/mcpwire.git
cd mcpwire
pip install -e .
Configuration (mcp.json)
The client can be configured using an mcp.json file. This allows you to define multiple server environments and easily switch between them using MCPClient.from_config().
The client searches for mcp.json in the following order:
- An explicit path provided to
MCPClient.from_config(config_path=...). - mcp.json in the current working directory.
.mcp.jsonin your home directory (~/.mcp.json)..config/mcp/mcp.jsonin your home directory (~/.config/mcp/mcp.json).
Example mcp.json:
{
"default_server": "local",
"servers": {
"local": {
"base_url": "http://localhost:8000",
"api_key": null,
"timeout": 60,
"transport": "sse",
"description": "Local development server (e.g., fast-agent)",
"resources": {
"enabled": true,
"auto_subscribe": ["file:///workspace/shared/*"],
"default_templates": [
{
"uri_template": "file:///workspace/{path}",
"name": "Workspace File",
"description": "Access files in the workspace directory"
},
{
"uri_template": "db://customers/{customer_id}",
"name": "Customer Record",
"description": "Access customer data from the database",
"mime_type": "application/json"
}
]
}
},
"stdio_server": {
"transport": "stdio",
"command": "python",
"args": ["-m", "mcp.server.cli"],
"timeout": 60,
"description": "Local stdio server",
"resources": {
"enabled": false
}
},
"remote_dev": {
"base_url": "https://dev-mcp.example.com",
"api_key": "dev-secret-key-abc",
"timeout": 90,
"transport": "sse",
"description": "Remote development/testing server",
"resources": {
"enabled": true,
"auto_subscribe": ["screen://capture/*"],
"default_templates": [
{
"uri_template": "screen://capture/{timestamp}",
"name": "Screen Capture",
"description": "Access screenshots captured during browsing",
"mime_type": "image/png"
}
]
}
},
"production": {
"base_url": "https://mcp.example.com",
"api_key": "env:MCP_PROD_API_KEY",
"timeout": 120,
"transport": "sse",
"description": "Production MCP server"
}
}
}
default_server: (Optional) The key of the server to use if server_name is omitted in from_config().servers: Dictionary of named server configurations.base_url: (Required for http/sse) Server URL.api_key: (Optional) API key string, null, or "env:YOUR_ENV_VAR".timeout: (Optional) Request timeout in seconds (defaults to 60).transport: (Required) The transport protocol to use ("sse" or "stdio").command: (Required for stdio) The command to execute.args: (Required for stdio) Arguments for the command.description: (Optional) Description.resources: (Optional) Configuration for MCP resources.enabled: (Required) Whether resources are enabled for this server.auto_subscribe: (Optional) List of resource URI patterns to automatically subscribe to.default_templates: (Optional) List of resource templates that define dynamic resources.uri_template: (Required) URI template following RFC 6570.name: (Required) Human-readable name for this resource type.description: (Optional) Description of this resource type.mime_type: (Optional) MIME type for all matching resources.
Quickstart (Async)
import asyncio
import logging
from mcpwire import (
MCPClient,
MultiServerMCPClient,
StdioConnection,
SSEConnection,
MCPAPIError,
MCPConnectionError,
MCPDataError
)
logging.basicConfig(level=logging.INFO)
async def main():
# --- Option 1: Initialize directly ---
# Note: The official MCP library supports "sse" transport, not "http"
client = MCPClient(base_url="http://localhost:8000", transport="sse")
try:
# Initialize and use the client
async with client as mcp:
# Get server metadata
metadata = await mcp.get_server_metadata()
logging.info(f"Connected to MCP server: {metadata.name} v{metadata.version}")
# List available tools
tools = await mcp.list_tools()
for tool in tools:
logging.info(f"Available tool: {tool.name}")
# Get a prompt
prompt_messages = await mcp.get_prompt("hello_world", {})
for msg in prompt_messages:
logging.info(f"[{msg.type}] {msg.content}")
# Call a tool
tool_result = await mcp.call_tool("search", {"query": "MCP protocol"})
logging.info(f"Tool result: {tool_result}")
# Working with resources
# List available resources
resources_result = await mcp.list_resources()
logging.info(f"Found {len(resources_result.resources)} resources and {len(resources_result.templates)} templates")
# Display available resources
for resource in resources_result.resources:
logging.info(f"Resource: {resource.name} (URI: {resource.uri})")
# Read a resource content (if any resources available)
if resources_result.resources:
resource_uri = resources_result.resources[0].uri
content = await mcp.read_resource(resource_uri)
logging.info(f"Read resource content with URI: {resource_uri}")
# Subscribe to resource updates
await mcp.subscribe_to_resource(resource_uri)
logging.info(f"Subscribed to resource: {resource_uri}")
# Later, unsubscribe when no longer needed
await mcp.unsubscribe_from_resource(resource_uri)
logging.info(f"Unsubscribed from resource: {resource_uri}")
except (MCPAPIError, MCPConnectionError, MCPDataError) as e:
logging.error(f"MCP Error: {e}")
except Exception as e:
logging.error(f"An unexpected error occurred: {e}", exc_info=True)
# --- Option 2: Initialize from mcp.json ---
try:
# Load 'local' config from mcp.json
client_from_config = MCPClient.from_config(server_name="local")
async with client_from_config as mcp:
logging.info(f"Using server configured as 'local'")
tools = await mcp.list_tools()
logging.info(f"Found {len(tools)} tools")
except (FileNotFoundError, ValueError, KeyError, MCPDataError) as e:
logging.error(f"Configuration Error: {e}")
except Exception as e:
logging.error(f"An unexpected error occurred: {e}", exc_info=True)
# --- Option 3: Use MultiServerMCPClient ---
try:
# Connect to multiple servers
connections = {
"math": {
"transport": "stdio",
"command": "python",
"args": ["-m", "mcp.server.cli"],
},
"web": {
"transport": "sse",
"url": "http://localhost:8000/sse",
}
}
async with MultiServerMCPClient(connections) as multi_client:
# Get tools from all servers
all_tools = multi_client.get_tools()
logging.info(f"Total tools from all servers: {len(all_tools)}")
# Get a prompt from a specific server
messages = await multi_client.get_prompt("math", "calculate", {"expression": "2+2"})
for msg in messages:
logging.info(f"Message: {msg.content}")
# Working with resources from multiple servers
# List resources from a specific server
math_resources = await multi_client.list_resources("math")
logging.info(f"Math server has {len(math_resources.resources)} resources")
# Read a resource from a specific server (if any available)
if math_resources.resources:
math_resource_uri = math_resources.resources[0].uri
math_content = await multi_client.read_resource("math", math_resource_uri)
logging.info(f"Read math resource: {math_resource_uri}")
# Subscribe and unsubscribe to resource updates
await multi_client.subscribe_to_resource("math", math_resource_uri)
await multi_client.unsubscribe_from_resource("math", math_resource_uri)
except Exception as e:
logging.error(f"MultiServerMCPClient error: {e}", exc_info=True)
# Run the async main function
if __name__ == "__main__":
asyncio.run(main())
Migration from v0.4.1 to v0.5.1
Version 0.5.1 adds comprehensive support for MCP Resources:
- Resource API: New methods for working with resources (
list_resources(),read_resource(), etc.) - Configuration: Added resource configuration options in mcp.json
- Resource Templates: Support for URI templates to access dynamic resources
- Resource Subscriptions: Subscribe to resource updates with
subscribe_to_resource() - MultiServer Resources: Access resources from multiple servers with MultiServerMCPClient
Development
- Clone:
git clone https://github.com/anukchat/mcpwire.git - Setup Env: Create and activate a virtual environment
- Install Dev Dependencies:
pip install -e ".[dev]"
Contributing
Contributions are welcome! Here's how you can contribute to the project:
- Fork the Repository: Create your own fork of the project
- Create a Branch: Create a branch for your changes
- Make Changes: Implement your changes, following the coding style
- Run Tests: Make sure the tests pass by running
pytest - Submit a Pull Request: Open a PR with a clear description of your changes
Testing
Run the tests using pytest:
# Install test dependencies
pip install -e ".[dev]"
# Run tests
pytest
Continuous Integration
This project uses GitHub Actions for continuous integration:
- Tests: Automatically run on each push and pull request to main/master branches
- Builds: Package is built and validated to ensure it will install correctly
- Publishing: When a new version tag is pushed (e.g., v0.4.1), the package is automatically published to PyPI
Creating a Release
To release a new version:
- Update the version in
pyproject.tomlandmcpwire/__init__.py - Update the CHANGELOG.md file with the changes
- Commit the changes and push to GitHub
- Create and push a new tag that matches the version:
git tag -a v0.5.1 -m "Release v0.5.1"
git push origin v0.5.1
The GitHub Actions workflow will automatically build and publish the package to PyPI.
License
This project is licensed under the MIT License - see the LICENSE file for details.
Troubleshooting Resources
When working with MCP resources, the following issues might occur:
Resource Errors
-
Server Not Found: When using
MultiServerMCPClient, ensure the server name exists before accessing resources.# This will raise ValueError if "math_server" is not connected resources = await multi_client.list_resources("math_server") -
Resource Not Found: Ensure the resource URI is valid and accessible.
try: resource = await client.read_resource("file:///nonexistent/file.txt") except MCPAPIError as e: print(f"Resource error: {e}") -
Binary Resource Handling: When dealing with binary resources (images, PDFs, etc.):
import base64 # Read a binary resource response = await client.read_resource("image:///logo.png") if response.contents[0].blob: # Decode base64 data binary_data = base64.b64decode(response.contents[0].blob) with open("logo.png", "wb") as f: f.write(binary_data) -
Template Parameters: When using templates, ensure all required parameters are provided:
# For a template like "file:///workspace/{path}" file_uri = "file:///workspace/documents/report.pdf" # Provide the complete URI -
Subscription Leaks: Always unsubscribe from resources when no longer needed:
try: await client.subscribe_to_resource("screen://capture/latest") # Do work with the resource... finally: # Ensure unsubscribe happens even if errors occur await client.unsubscribe_from_resource("screen://capture/latest")
Common Configuration Issues
-
Missing Resource Configuration: Ensure the server has resources enabled:
"resources": { "enabled": true } -
Invalid Template Format: Resource templates must follow RFC 6570 URI template format:
"uri_template": "file:///workspace/{path}", // Correct "uri_template": "file:///workspace/<path>", // Incorrect -
Incorrect MIME Types: Use standard MIME types for resources to ensure proper handling:
"mime_type": "application/json" // For JSON data "mime_type": "text/plain" // For text files "mime_type": "image/png" // For PNG images "mime_type": "application/pdf" // For PDF documents
