Building Your First AI Agent with MCP

# Create project directory uv init my-first-mcp-agent cd my-first-mcp-agent # Create virtual environment uv venv source .venv/bin/activate # On Windows: .venv\Scripts\activate # Install MCP SDK with development tools uv add "mcp[cli]" httpx pydantic redis python-dotenv

# hello_mcp.py - Your first MCP server from mcp.server.fastmcp import FastMCP import asyncio # Create server instance with descriptive name mcp = FastMCP("Hello World AI Agent") @mcp.tool() def greet_user(name: str) -> str: """Greet a user by name with a friendly message""" return f"Hello, {name}! 👋 I'm your first MCP agent, ready to help!" @mcp.tool() def get_server_stats() -> dict: """Provide basic server statistics and information""" return { "server_name": "Hello World AI Agent", "version": "1.0.0", "status": "active", "capabilities": ["greeting", "information"] } @mcp.resource("info://server") def server_info() -> str: """Provide comprehensive server information""" return """ Hello World MCP Server - Your First AI Agent! This server demonstrates basic MCP concepts: - Tool registration and execution - Resource provision - JSON-RPC communication protocol Available tools: - greet_user(name): Personal greeting functionality - get_server_stats(): Server status and capabilities Ready to build more advanced agents! """ if __name__ == "__main__": # Run the server asyncio.run(mcp.run())

# weather_agent.py - Practical weather intelligence agent from mcp.server.fastmcp import FastMCP import httpx import asyncio import os from typing import Dict, Any, Optional from pydantic import BaseModel, validator from datetime import datetime import logging # Setup logging for debugging and monitoring logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Data validation models class WeatherQuery(BaseModel): city: str country: str = "US" units: str = "metric" @validator('city') def validate_city(cls, v): if not v or len(v) > 100: raise ValueError('City name is required and must be under 100 characters') return v.strip() # Initialize MCP server mcp = FastMCP("Weather Intelligence Agent") # Configuration API_KEY = os.getenv("OPENWEATHER_API_KEY", "your-api-key-here") BASE_URL = "https://api.openweathermap.org/data/2.5" @mcp.tool() async def get_current_weather(city: str, country: str = "US") -> Dict[str, Any]: """Get current weather conditions for a specified city Args: city: Name of the city (e.g., 'San Francisco') country: Country code (e.g., 'US', 'UK', 'CA') Returns: Dict containing weather data or error information """ try: # Validate input query = WeatherQuery(city=city, country=country) # Build API request url = f"{BASE_URL}/weather" params = { "q": f"{query.city},{query.country}", "appid": API_KEY, "units": query.units } # Make async API call with timeout async with httpx.AsyncClient() as client: response = await client.get( url, params=params, timeout=10.0 ) # Handle successful response if response.status_code == 200: data = response.json() logger.info(f"Successfully retrieved weather for {city}, {country}") return { "success": True, "city": data["name"], "country": data["sys"]["country"], "temperature": data["main"]["temp"], "feels_like": data["main"]["feels_like"], "description": data["weather"][0]["description"].title(), "humidity": data["main"]["humidity"], "pressure": data["main"]["pressure"], "wind_speed": data.get("wind", {}).get("speed", 0), "timestamp": datetime.now().isoformat() } else: logger.error(f"API request failed: {response.status_code}") return { "success": False, "error": f"Weather API returned status {response.status_code}", "message": "Unable to fetch weather data. Please check city name and try again." } except ValueError as e: logger.error(f"Validation error: {e}") return { "success": False, "error": "validation_error", "message": str(e) } except httpx.TimeoutException: logger.error(f"Timeout fetching weather for {city}") return { "success": False, "error": "timeout", "message": "Request timed out. Please try again later." } except Exception as e: logger.exception(f"Unexpected error fetching weather: {e}") return { "success": False, "error": "internal_error", "message": "An internal error occurred. Please try again." } @mcp.tool() async def get_weather_forecast(city: str, days: int = 5) -> Dict[str, Any]: """Get weather forecast for the next few days Args: city: Name of the city days: Number of days to forecast (1-5) Returns: Dict containing forecast data """ try: # Validate inputs if days < 1 or days > 5: return { "success": False, "error": "validation_error", "message": "Days must be between 1 and 5" } # API call for forecast data url = f"{BASE_URL}/forecast" params = { "q": city, "appid": API_KEY, "units": "metric", "cnt": days * 8 # 8 forecasts per day (every 3 hours) } async with httpx.AsyncClient() as client: response = await client.get(url, params=params, timeout=10.0) if response.status_code == 200: data = response.json() return { "success": True, "city": data["city"]["name"], "forecast_count": len(data["list"]), "forecasts": [ { "datetime": item["dt_txt"], "temperature": item["main"]["temp"], "description": item["weather"][0]["description"].title(), "humidity": item["main"]["humidity"] } for item in data["list"][:days*2] # Limit to reasonable amount ] } else: return { "success": False, "error": f"API error: {response.status_code}" } except Exception as e: logger.exception(f"Error fetching forecast: {e}") return { "success": False, "error": "internal_error", "message": str(e) } @mcp.resource("weather://cities/{city}") async def city_weather_info(city: str) -> str: """Get comprehensive weather information for a city as a formatted resource""" weather_data = await get_current_weather(city) if weather_data["success"]: return f""" Weather Report for {weather_data['city']}, {weather_data['country']} Generated: {weather_data['timestamp']} 🌡️ Temperature: {weather_data['temperature']}°C (feels like {weather_data['feels_like']}°C) 🌤️ Conditions: {weather_data['description']} 💧 Humidity: {weather_data['humidity']}% 🗜️ Pressure: {weather_data['pressure']} hPa 💨 Wind Speed: {weather_data['wind_speed']} m/s This data is provided by OpenWeatherMap API and updated in real-time. """ else: return f"Weather data unavailable for {city}. Error: {weather_data.get('message', 'Unknown error')}" if __name__ == "__main__": # Run the weather intelligence agent print("🌤️ Starting Weather Intelligence Agent...") print("Make sure to set your OPENWEATHER_API_KEY environment variable!") asyncio.run(mcp.run())

# production_weather_agent.py - Enterprise-ready MCP server from mcp.server.fastmcp import FastMCP import httpx import asyncio import redis.asyncio as redis import json import os from typing import Dict, Any, Optional from pydantic import BaseModel, validator from datetime import datetime, timedelta from enum import Enum import logging import time from functools import wraps # Enhanced logging configuration logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s', handlers=[ logging.FileHandler('weather_agent.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) # Error categorization class ErrorCode(Enum): VALIDATION_ERROR = "VALIDATION_ERROR" API_ERROR = "API_ERROR" RATE_LIMIT = "RATE_LIMIT" CACHE_ERROR = "CACHE_ERROR" NETWORK_ERROR = "NETWORK_ERROR" INTERNAL_ERROR = "INTERNAL_ERROR" # Enhanced validation models class WeatherQuery(BaseModel): city: str country: str = "US" units: str = "metric" @validator('city') def validate_city(cls, v): if not v or len(v.strip()) == 0: raise ValueError('City name is required') if len(v) > 100: raise ValueError('City name must be under 100 characters') # Basic sanitization sanitized = ''.join(c for c in v if c.isalnum() or c in ' -') return sanitized.strip() @validator('units') def validate_units(cls, v): valid_units = ['metric', 'imperial', 'kelvin'] if v not in valid_units: raise ValueError(f'Units must be one of {valid_units}') return v # Production MCP server with caching class ProductionWeatherAgent: def __init__(self): self.mcp = FastMCP("Production Weather Intelligence Agent") self.api_key = os.getenv("OPENWEATHER_API_KEY") self.redis_client: Optional[redis.Redis] = None self.http_client: Optional[httpx.AsyncClient] = None self.request_count = 0 self.start_time = time.time() # Rate limiting configuration self.rate_limit_requests = 100 # requests per hour self.rate_limit_window = 3600 # 1 hour self._register_tools() async def initialize(self): """Initialize async components""" try: # Initialize Redis connection redis_url = os.getenv("REDIS_URL", "redis://localhost:6379") self.redis_client = redis.from_url(redis_url, decode_responses=True) await self.redis_client.ping() logger.info("Redis connection established") except Exception as e: logger.warning(f"Redis connection failed: {e}. Continuing without caching.") self.redis_client = None # Initialize HTTP client with connection pooling timeout = httpx.Timeout(10.0, connect=5.0) limits = httpx.Limits(max_keepalive_connections=5, max_connections=10) self.http_client = httpx.AsyncClient(timeout=timeout, limits=limits) def rate_limit(self, func): """Decorator for rate limiting""" @wraps(func) async def wrapper(*args, **kwargs): if await self._check_rate_limit(): return { "success": False, "error_code": ErrorCode.RATE_LIMIT.value, "message": "Rate limit exceeded. Please try again later." } return await func(*args, **kwargs) return wrapper async def _check_rate_limit(self) -> bool: """Check if rate limit is exceeded""" current_time = time.time() if current_time - self.start_time > self.rate_limit_window: self.request_count = 0 self.start_time = current_time self.request_count += 1 return self.request_count > self.rate_limit_requests async def _get_cached_data(self, cache_key: str) -> Optional[dict]: """Retrieve data from cache""" if not self.redis_client: return None try: cached_data = await self.redis_client.get(cache_key) if cached_data: return json.loads(cached_data) except Exception as e: logger.error(f"Cache retrieval error: {e}") return None async def _set_cached_data(self, cache_key: str, data: dict, ttl: int = 600): """Store data in cache""" if not self.redis_client: return try: await self.redis_client.setex(cache_key, ttl, json.dumps(data)) except Exception as e: logger.error(f"Cache storage error: {e}") def _register_tools(self): """Register all MCP tools""" @self.mcp.tool() @self.rate_limit async def get_current_weather_cached(city: str, country: str = "US") -> Dict[str, Any]: """Get current weather with intelligent caching""" return await self._get_weather_with_cache(city, country) @self.mcp.tool() async def get_agent_health() -> Dict[str, Any]: """Get agent health and performance metrics""" return { "status": "healthy", "uptime_seconds": time.time() - self.start_time, "requests_processed": self.request_count, "cache_available": self.redis_client is not None, "api_key_configured": bool(self.api_key and self.api_key != "your-api-key-here"), "timestamp": datetime.now().isoformat() } async def _get_weather_with_cache(self, city: str, country: str) -> Dict[str, Any]: """Get weather data with caching logic""" try: # Validate input query = WeatherQuery(city=city, country=country) cache_key = f"weather:{query.city}:{query.country}" # Try cache first cached_result = await self._get_cached_data(cache_key) if cached_result: logger.info(f"Cache hit for {city}, {country}") cached_result["cache_hit"] = True return cached_result # Fetch from API weather_data = await self._fetch_weather_api(query) # Cache successful results if weather_data["success"]: weather_data["cache_hit"] = False await self._set_cached_data(cache_key, weather_data, 600) # 10 minutes return weather_data except ValueError as e: logger.error(f"Validation error: {e}") return { "success": False, "error_code": ErrorCode.VALIDATION_ERROR.value, "message": str(e) } except Exception as e: logger.exception(f"Unexpected error: {e}") return { "success": False, "error_code": ErrorCode.INTERNAL_ERROR.value, "message": "Internal server error" } async def _fetch_weather_api(self, query: WeatherQuery) -> Dict[str, Any]: """Fetch weather data from OpenWeatherMap API""" if not self.api_key or self.api_key == "your-api-key-here": return { "success": False, "error_code": ErrorCode.API_ERROR.value, "message": "API key not configured" } url = "https://api.openweathermap.org/data/2.5/weather" params = { "q": f"{query.city},{query.country}", "appid": self.api_key, "units": query.units } try: response = await self.http_client.get(url, params=params) if response.status_code == 200: data = response.json() logger.info(f"API success for {query.city}, {query.country}") return { "success": True, "city": data["name"], "country": data["sys"]["country"], "temperature": data["main"]["temp"], "feels_like": data["main"]["feels_like"], "description": data["weather"][0]["description"].title(), "humidity": data["main"]["humidity"], "pressure": data["main"]["pressure"], "wind_speed": data.get("wind", {}).get("speed", 0), "timestamp": datetime.now().isoformat() } else: logger.error(f"API error {response.status_code}: {response.text}") return { "success": False, "error_code": ErrorCode.API_ERROR.value, "message": f"Weather API returned status {response.status_code}" } except httpx.TimeoutException: logger.error(f"Timeout fetching weather for {query.city}") return { "success": False, "error_code": ErrorCode.NETWORK_ERROR.value, "message": "Request timed out" } except Exception as e: logger.exception(f"Network error: {e}") return { "success": False, "error_code": ErrorCode.NETWORK_ERROR.value, "message": "Network error occurred" } async def run(self): """Run the production weather agent""" await self.initialize() await self.mcp.run() async def cleanup(self): """Cleanup resources""" if self.http_client: await self.http_client.aclose() if self.redis_client: await self.redis_client.close() # Main execution async def main(): agent = ProductionWeatherAgent() try: print("🚀 Starting Production Weather Intelligence Agent...") await agent.run() finally: await agent.cleanup() if __name__ == "__main__": asyncio.run(main())

// ~/.claude/claude_desktop_config.json { "mcpServers": { "weather-agent": { "command": "/absolute/path/to/python", "args": ["/absolute/path/to/production_weather_agent.py"], "env": { "OPENWEATHER_API_KEY": "your_actual_api_key_here", "REDIS_URL": "redis://localhost:6379" } } } }

@mcp.tool() async def comprehensive_weather_report(city: str) -> Dict[str, Any]: """Generate comprehensive weather report using multiple data sources""" # Coordinate multiple tools for complete analysis current_weather = await get_current_weather_cached(city) forecast = await get_weather_forecast(city, days=3) # Combine and analyze data report = { "city": city, "current": current_weather, "forecast": forecast, "analysis": generate_weather_insights(current_weather, forecast), "recommendations": generate_activity_recommendations(current_weather) } return report

def register_weather_tools(mcp_server, config): """Dynamically register tools based on available APIs""" if config.get("openweather_api_key"): mcp_server.register_tool("current_weather", get_current_weather) mcp_server.register_tool("forecast", get_weather_forecast) if config.get("weather_gov_access"): mcp_server.register_tool("us_weather", get_us_weather_service) if config.get("accuweather_key"): mcp_server.register_tool("detailed_forecast", get_accuweather_data)