Building a Lightweight MCP Client Using Python and HTTP Libraries
Welcome to this hands-on tutorial where we will be building a lightweight MCP client using Python and popular HTTP libraries. The aim is to give you the tools and understanding necessary to integrate MCP capabilities into your applications seamlessly. Whether you're a beginner or an experienced developer, this tutorial will offer practical insights and code to enhance your project.
Prerequisites
- Intermediate knowledge of Python programming
- Basic understanding of HTTP requests and responses
- Python environment setup (Python 3.7 or later)
Step 1: Setting Up Your Development Environment
We’ll start by setting up our environment and installing the necessary libraries. We will use the requests library for HTTP requests and json for handling JSON data.
# Create a new directory for the project
mkdir mcp-client
cd mcp-client
# Create a new virtual environment
python3 -m venv venv
source venv/bin/activate # On Windows use `venv\Scripts\activate`
# Install required libraries
pip install requestsStep 2: Building the MCP Client
Let’s define the skeleton of our MCP client. We will create a class named MCPClient which will handle making requests to and parsing responses from the MCP server.
import requests
class MCPClient:
def __init__(self, base_url):
self.base_url = base_url
def send_request(self, endpoint, data):
url = f"{self.base_url}/{endpoint}"
try:
response = requests.post(url, json=data)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as http_err:
print(f"HTTP error occurred: {http_err}")
except Exception as err:
print(f"Other error occurred: {err}")
return NoneThis basic client initializes with a base URL and provides a send_request method for interacting with the MCP server endpoints.
Step 3: Sending and Handling Requests
Now, let's write a function that uses our MCPClient to send a request to an MCP server. We'll handle potential errors and ensure that the response data is correctly parsed.
def interact_with_mcp():
client = MCPClient(base_url="https://example-mcp-server.com/api")
endpoint = "process"
payload = {"command": "run-analysis", "parameters": {"param1": "value1"}}
result = client.send_request(endpoint, payload)
if result:
print("Successfully received response from MCP server:")
print(result)
else:
print("Failed to retrieve valid response.")
interact_with_mcp()This function creates a new MCPClient instance, sets up the necessary endpoint and data payload, sends the request, and prints the result.
Step 4: Enhancing the Client with Custom Headers
To handle scenarios where authentication or custom headers are needed, let's enhance our client to accept additional headers.
class EnhancedMCPClient(MCPClient):
def send_request(self, endpoint, data, headers=None):
url = f"{self.base_url}/{endpoint}"
try:
response = requests.post(url, json=data, headers=headers)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as http_err:
print(f"HTTP error occurred: {http_err}")
except Exception as err:
print(f"Other error occurred: {err}")
return NoneStep 5: Adding JWT Authentication
In a real-world application, you'll likely need to authenticate your requests using a JSON Web Token (JWT). Here's how you can modify the EnhancedMCPClient to include JWT authentication:
import jwt
class AuthenticatedMCPClient(EnhancedMCPClient):
def __init__(self, base_url, secret_key):
super().__init__(base_url)
self.secret_key = secret_key
def generate_jwt(self, payload):
return jwt.encode(payload, self.secret_key, algorithm="HS256")
def send_request(self, endpoint, data, headers=None):
if headers is None:
headers = {}
jwt_token = self.generate_jwt({"iss": "your-issuer", "exp": 1609459200})
headers["Authorization"] = f"Bearer {jwt_token}"
return super().send_request(endpoint, data, headers)Step 6: Handling Rate Limiting
If your requests are being throttled, you're likely hitting a rate limit imposed by the MCP server. You can handle this by checking response headers for rate limit info and dynamically adjusting your request rate.
class RateLimitedMCPClient(EnhancedMCPClient):
def send_request(self, endpoint, data, headers=None):
response = super().send_request(endpoint, data, headers)
if response is not None:
rate_limit_remaining = response.headers.get("X-RateLimit-Remaining")
if rate_limit_remaining is not None and int(rate_limit_remaining) < 10:
print("Rate limit is low, slowing down requests")
time.sleep(1) # slow down requests
return responseCommon Pitfall: Rate Limiting
If your requests are being throttled, you're likely hitting a rate limit imposed by the MCP server. You can handle this by checking response headers for rate limit info and dynamically adjusting your request rate.
Performance Tip: Asynchronous Requests
If you need to perform multiple requests efficiently, consider using asyncio along with an asynchronous HTTP client like aiohttp. This can significantly improve performance by handling I/O-bound tasks concurrently.
import asyncio
import aiohttp
class AsyncMCPClient:
def __init__(self, base_url):
self.base_url = base_url
async def send_request(self, endpoint, data):
async with aiohttp.ClientSession() as session:
async with session.post(f"{self.base_url}/{endpoint}", json=data) as response:
return await response.json()
async def main():
client = AsyncMCPClient("https://example-mcp-server.com/api")
tasks = []
for i in range(10):
task = asyncio.create_task(client.send_request("process", {"command": "run-analysis", "parameters": {"param1": "value1"}}))
tasks.append(task)
results = await asyncio.gather(*tasks)
for result in results:
print(result)
asyncio.run(main())Advanced Error Handling
To improve the robustness of our client, let's implement advanced error handling. We'll catch specific exceptions and provide informative error messages.
class RobustMCPClient(EnhancedMCPClient):
def send_request(self, endpoint, data, headers=None):
try:
response = super().send_request(endpoint, data, headers)
if response is None:
raise Exception("Failed to retrieve a valid response")
return response
except requests.exceptions.ConnectionError as conn_err:
print(f"Connection error occurred: {conn_err}")
except requests.exceptions.Timeout as timeout_err:
print(f"Timeout error occurred: {timeout_err}")
except Exception as err:
print(f"An error occurred: {err}")
return NoneWith this advanced error handling, our client is more robust and provides better error messages.
Using a Retry Mechanism
To handle transient errors, we can implement a retry mechanism. We'll use the tenacity library to retry failed requests.
import tenacity
class RetryingMCPClient(EnhancedMCPClient):
@tenacity.retry(wait=tenacity.wait_exponential(multiplier=1, min=4, max=10))
def send_request(self, endpoint, data, headers=None):
return super().send_request(endpoint, data, headers)With this retry mechanism, our client will automatically retry failed requests, improving its overall reliability.
Monitoring Request Performance
To monitor the performance of our client, we can use a library like prometheus-client to track metrics such as request latency and success rate.
from prometheus_client import Counter, Gauge
class MonitoringMCPClient(EnhancedMCPClient):
def __init__(self, base_url):
super().__init__(base_url)
self.request_counter = Counter("mcp_requests", "Number of MCP requests")
self.request_latency_gauge = Gauge("mcp_request_latency", "MCP request latency")
def send_request(self, endpoint, data, headers=None):
start_time = time.time()
response = super().send_request(endpoint, data, headers)
end_time = time.time()
self.request_counter.inc()
self.request_latency_gauge.set(end_time - start_time)
return responseWith this monitoring capability, we can track the performance of our client and identify areas for improvement.