weather_agent_cli.py•7 kB
# weather_agent_cli.py
import httpx
import asyncio
import json
import sys # To exit the script
# --- Configuration for the Agent ---
MCP_SERVER_BASE_URL = "http://localhost:8000"
WEATHER_MCP_ENDPOINT = f"{MCP_SERVER_BASE_URL}/mcp/weather"
# Default MCP parameters
DEFAULT_PROTOCOL_VERSION = "1.0"
DEFAULT_TOOL_ID = "weather_tool"
DEFAULT_METHOD = "get_current_weather"
async def call_mcp_weather_server(location: str) -> dict | None:
"""
Calls the MCP Weather Server to get weather for a given location.
Returns the parsed JSON response or None if an error occurs before getting a JSON response.
"""
mcp_payload = {
"protocol_version": DEFAULT_PROTOCOL_VERSION,
"tool_id": DEFAULT_TOOL_ID,
"method": DEFAULT_METHOD,
"parameters": {
"location": location
}
}
try:
async with httpx.AsyncClient(timeout=10.0) as client:
print(f" [Agent making request to MCP: {location}]") # Debug print
response = await client.post(WEATHER_MCP_ENDPOINT, json=mcp_payload)
response.raise_for_status() # Will raise an exception for 4xx/5xx an external API would,
# but our MCP server returns 200 with status in payload.
# So, this might not be strictly necessary if server always 200s.
# However, good for network level errors before MCP payload.
return response.json() # Assuming MCP server always returns JSON
except httpx.HTTPStatusError as e:
print(f" [Agent Error] HTTP error calling MCP Server: {e.response.status_code} - {e.response.text}")
return {"status": "error", "error_message": f"HTTP error {e.response.status_code} when contacting MCP server."}
except httpx.ConnectError:
print(f" [Agent Error] Could not connect to MCP Server at {WEATHER_MCP_ENDPOINT}.")
print(f" Please ensure the MCP server is running.")
return {"status": "error", "error_message": "Agent could not connect to the MCP server."}
except httpx.TimeoutException:
print(f" [Agent Error] Request to MCP Server timed out.")
return {"status": "error", "error_message": "Request to MCP server timed out."}
except json.JSONDecodeError:
print(f" [Agent Error] Could not decode JSON response from MCP server.")
return {"status": "error", "error_message": "Invalid JSON response from MCP server."}
except Exception as e:
print(f" [Agent Error] An unexpected error occurred while calling MCP server: {e}")
return {"status": "error", "error_message": f"Unexpected agent error: {e}"}
async def process_user_input(user_input: str) -> str:
"""
Processes the user input, decides if it's a weather request,
calls the MCP server if needed, and formats the response.
"""
lower_input = user_input.lower()
location = None
if lower_input.startswith("weather in "):
location = user_input[len("weather in "):].strip()
elif lower_input.startswith("weather "):
location = user_input[len("weather "):].strip()
elif lower_input == "quit" or lower_input == "exit":
return "quit"
if location:
if not location: # If location ended up being an empty string after stripping
return "WeatherAgent: Please specify a location. For example: 'weather in London'."
mcp_response = await call_mcp_weather_server(location)
if mcp_response:
if mcp_response.get("status") == "success":
data = mcp_response.get("data")
if data:
# Using f-string for multi-line output and clarity
response_string = (
f"WeatherAgent: Weather in {data.get('location', 'N/A')}:\n"
f" Condition: {data.get('condition', 'N/A')} ({data.get('description', 'N/A')})\n"
f" Temperature: {data.get('temperature_celsius', 'N/A')}°C "
f"({data.get('temperature_fahrenheit', 'N/A')}°F)\n"
f" Humidity: {data.get('humidity_percent', 'N/A')}%\n"
f" Wind: {data.get('wind_kph', 'N/A')} kph\n"
f" Pressure: {data.get('pressure_hpa', 'N/A')} hPa"
)
return response_string
else:
return "WeatherAgent: Successfully contacted server, but no weather data was returned."
elif mcp_response.get("status") == "error":
error_msg = mcp_response.get("error_message", "An unknown error occurred.")
return f"WeatherAgent: Sorry, I couldn't get the weather. Reason: {error_msg}"
else:
return "WeatherAgent: Received an unexpected response format from the MCP server."
else:
# Error messages are printed by call_mcp_weather_server in case of connection issues etc.
# This path handles if call_mcp_weather_server itself returns None unexpectedly.
return "WeatherAgent: Failed to get a response from the MCP server."
elif lower_input == "quit" or lower_input == "exit":
return "quit" # Should be caught by the first check, but good to have
else:
return "WeatherAgent: I can only provide weather information. Try 'weather in [city]' or 'weather [city]' (e.g., 'weather in London'). Type 'quit' to exit."
async def main_loop():
"""
Main loop for the WeatherAgent CLI.
"""
print("WeatherAgent CLI Initialized (type 'quit' or 'exit' to stop)")
print("Example: weather in London")
print("-" * 30)
while True:
try:
user_input = input("You> ")
if not user_input.strip(): # Handle empty input
continue
agent_response = await process_user_input(user_input)
if agent_response == "quit":
print("WeatherAgent: Exiting. Goodbye!")
break
else:
print(agent_response)
except KeyboardInterrupt: # Handle Ctrl+C
print("\nWeatherAgent: Exiting due to user interrupt. Goodbye!")
break
except Exception as e: # Catch any other loop-level errors
print(f"WeatherAgent: A critical error occurred in the agent loop: {e}")
# Optionally, decide if you want to break the loop or continue
# break
if __name__ == "__main__":
# Ensure the MCP server is running: uvicorn app.main:app --reload --port 8000
# Then run this agent: python weather_agent_cli.py
try:
asyncio.run(main_loop())
except KeyboardInterrupt:
# This handles Ctrl+C if it happens right at startup or during cleanup
# (though the loop itself also has a KeyboardInterrupt handler)
print("\nWeatherAgent: Shutting down...")