test_mcp_client.py•4.67 kB
import httpx
import asyncio
import json # For pretty-printing JSON output
# --- Configuration for the Client ---
MCP_SERVER_BASE_URL = "http://localhost:8000" # Base URL of your running MCP server
WEATHER_MCP_ENDPOINT = f"{MCP_SERVER_BASE_URL}/mcp/weather"
# Default MCP parameters for weather requests
DEFAULT_PROTOCOL_VERSION = "1.0"
DEFAULT_TOOL_ID = "weather_tool"
DEFAULT_METHOD = "get_current_weather"
async def request_weather(location: str,
tool_id: str = DEFAULT_TOOL_ID,
method: str = DEFAULT_METHOD,
protocol_version: str = DEFAULT_PROTOCOL_VERSION) -> None:
"""
Sends a request to the MCP Weather Server to get weather for a given location
and prints the server's response.
"""
mcp_payload = {
"protocol_version": protocol_version,
"tool_id": tool_id,
"method": method,
"parameters": {
"location": location
}
}
print(f"\n>>> Testing weather request for: '{location}'")
print(f" Endpoint: {WEATHER_MCP_ENDPOINT}")
print(f" Payload: {json.dumps(mcp_payload, indent=2)}")
try:
async with httpx.AsyncClient(timeout=10.0) as client: # 10-second timeout
response = await client.post(WEATHER_MCP_ENDPOINT, json=mcp_payload)
print(f"\n<<< Response for '{location}':")
print(f" Status Code: {response.status_code}")
try:
response_json = response.json()
print(f" Response Body (JSON):\n{json.dumps(response_json, indent=2)}")
# Basic check of MCP response structure
if "status" in response_json:
if response_json["status"] == "success" and "data" in response_json:
print(f" MCP Status: SUCCESS - Weather data received for {response_json['data'].get('location', location)}.")
elif response_json["status"] == "error" and "error_message" in response_json:
print(f" MCP Status: ERROR - {response_json['error_message']}")
else:
print(" MCP Status: UNKNOWN - Response format unexpected.")
else:
print(" WARNING: 'status' field missing in MCP response.")
except json.JSONDecodeError:
print(" Error: Could not decode JSON response from server.")
print(f" Raw Response Text:\n{response.text}")
except httpx.ConnectError as e:
print(f"\n CONNECTION ERROR: Could not connect to MCP Server at {WEATHER_MCP_ENDPOINT}.")
print(f" Details: {e}")
print(f" Please ensure the MCP server is running via 'uvicorn app.main:app --reload --port 8000'")
except httpx.TimeoutException as e:
print(f"\n TIMEOUT ERROR: The request to {WEATHER_MCP_ENDPOINT} timed out.")
print(f" Details: {e}")
except Exception as e:
print(f"\n UNEXPECTED ERROR during request for '{location}':")
print(f" Details: {type(e).__name__} - {e}")
print("-" * 50)
async def main():
"""
Main function to run a series of test requests.
"""
print("Starting MCP Weather Client Tests...")
print(f"Targeting MCP Server at: {MCP_SERVER_BASE_URL}")
print("=" * 50)
# Test Case 1: Successful weather request
await request_weather("New York, US")
# Test Case 2: Location likely to be found but different format
await request_weather("Tokyo")
# Test Case 3: Location not found by OpenWeatherMap
await request_weather("NonExistentCity123abc")
# Test Case 4: Request with an incorrect tool_id (testing server-side MCP validation)
await request_weather("London,UK", tool_id="wrong_tool")
# Test Case 5: Request with an incorrect method (testing server-side MCP validation)
await request_weather("Berlin,DE", method="do_something_else")
# Test Case 6: Empty location string (see how the server handles it)
# Depending on your OpenWeatherMap API behavior and server validation,
# this might result in an error from OpenWeatherMap or your service layer.
await request_weather("")
print("\nFinished MCP Weather Client Tests.")
print("=" * 50)
if __name__ == "__main__":
# Ensure the MCP server is running before executing this script.
# To run: python test_mcp_client.py
# Using asyncio.run() for Python 3.7+
asyncio.run(main())
# For older Python versions (e.g., 3.6), you might need:
# loop = asyncio.get_event_loop()
# try:
# loop.run_until_complete(main())
# finally:
# loop.close()