{ "cells": [ { "cell_type": "raw", "metadata": { "vscode": { "languageId": "raw" } }, "source": [ "# 🌐 API Integration with FastMCP\n", "\n", "Welcome to API integration with FastMCP! In this notebook, you'll learn how to build robust API integrations using modern MCP patterns and type-safe operations.\n", "\n", "## 🎯 Learning Objectives\n", "\n", "By the end of this notebook, you will:\n", "- Build type-safe API clients with FastMCP\n", "- Implement API resources and caching\n", "- Handle authentication and rate limiting\n", "- Use Pydantic for request/response validation\n", "- Apply MCP best practices for API integration\n", "\n", "## 🛠️ What You'll Build\n", "\n", "```python\n", "from mcp.server.fastmcp import FastMCP\n", "from pydantic import BaseModel\n", "from typing import Optional, List\n", "\n", "# Initialize FastMCP server\n", "mcp = FastMCP(\"api_tools\")\n", "\n", "class WeatherData(BaseModel):\n", " \"\"\"Weather data model\"\"\"\n", " location: str\n", " temperature: float\n", " conditions: str\n", " humidity: Optional[int]\n", "\n", "@mcp.resource(\"weather://{location}\")\n", "async def get_weather_data(location: str) -> WeatherData:\n", " \"\"\"Get cached weather data\"\"\"\n", " ...\n", "\n", "@mcp.tool()\n", "async def get_forecast(location: str) -> Dict:\n", " \"\"\"Get weather forecast\"\"\"\n", " ...\n", "\n", "@mcp.prompt()\n", "def error_prompt() -> str:\n", " \"\"\"Handle API errors\"\"\"\n", " return \"\"\"API request failed...\"\"\"\n", "```\n", "\n", "## 🔐 Modern API Patterns\n", "\n", "FastMCP provides robust API integration through:\n", "- **Type-safe requests/responses**\n", "- **Resource caching**\n", "- **Rate limiting**\n", "- **Error handling**\n", "- **Authentication management**\n", "\n", "## 📚 Table of Contents\n", "\n", "1. [FastMCP Setup](#fastmcp-setup)\n", "2. [API Resources](#api-resources)\n", "3. [API Tools](#api-tools)\n", "4. [Authentication](#authentication)\n", "5. [Error Handling](#error-handling)\n", "6. [Production Patterns](#production-patterns)\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# First, let's set up our FastMCP server with necessary imports\n", "from mcp.server.fastmcp import FastMCP\n", "import mcp.types as types\n", "from pydantic import BaseModel, Field\n", "from typing import Dict, List, Optional\n", "from datetime import datetime\n", "import httpx\n", "import asyncio\n", "from dataclasses import dataclass\n", "import json\n", "\n", "# Initialize our FastMCP server\n", "mcp = FastMCP(\"api_tools\")\n", "\n", "# API Configuration\n", "@dataclass\n", "class APIConfig:\n", " \"\"\"API configuration\"\"\"\n", " base_url: str = \"https://api.openweathermap.org/data/2.5\"\n", " api_key: str = \"demo_key\" # In production, use environment variables\n", " timeout: int = 10\n", " max_retries: int = 3\n", "\n", "config = APIConfig()\n", "\n", "# Data Models\n", "class Location(BaseModel):\n", " \"\"\"Location model\"\"\"\n", " name: str\n", " country: str\n", " latitude: float\n", " longitude: float\n", "\n", "class WeatherData(BaseModel):\n", " \"\"\"Weather data model\"\"\"\n", " location: Location\n", " temperature: float\n", " conditions: str\n", " humidity: Optional[int]\n", " wind_speed: float\n", " timestamp: datetime = Field(default_factory=datetime.now)\n", "\n", "class ForecastDay(BaseModel):\n", " \"\"\"Daily forecast model\"\"\"\n", " date: datetime\n", " temp_high: float\n", " temp_low: float\n", " conditions: str\n", " precipitation: float\n", "\n", "class Forecast(BaseModel):\n", " \"\"\"Forecast model\"\"\"\n", " location: Location\n", " days: List[ForecastDay]\n", " generated_at: datetime = Field(default_factory=datetime.now)\n", "\n", "# System prompts\n", "@mcp.prompt()\n", "def system_prompt() -> str:\n", " \"\"\"Define the API client's role\"\"\"\n", " return \"\"\"\n", " You are an API integration assistant that helps users access weather data safely.\n", " Always validate API requests and handle rate limits and errors appropriately.\n", " Provide clear error messages when API calls fail.\n", " \"\"\"\n", "\n", "@mcp.prompt()\n", "def error_prompt() -> str:\n", " \"\"\"Handle API errors\"\"\"\n", " return \"\"\"\n", " The API request failed.\n", " This could be due to:\n", " - Invalid API key\n", " - Rate limiting\n", " - Network issues\n", " - Invalid parameters\n", " Please check your request and try again.\n", " \"\"\"\n", "\n", "print(\"✅ FastMCP server initialized with API models and prompts!\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Let's implement our API resource provider\n", "@mcp.resource(\"weather://{location}\")\n", "async def get_weather_data(location: str) -> Dict:\n", " \"\"\"Get cached weather data for a location\"\"\"\n", " try:\n", " # In production, check cache first\n", " \n", " # Make API request\n", " async with httpx.AsyncClient() as client:\n", " response = await client.get(\n", " f\"{config.base_url}/weather\",\n", " params={\n", " \"q\": location,\n", " \"appid\": config.api_key,\n", " \"units\": \"metric\"\n", " },\n", " timeout=config.timeout\n", " )\n", " \n", " if response.status_code == 429:\n", " raise Exception(\"Rate limit exceeded\")\n", " \n", " response.raise_for_status()\n", " data = response.json()\n", " \n", " # Parse response into our model\n", " weather = WeatherData(\n", " location=Location(\n", " name=data[\"name\"],\n", " country=data[\"sys\"][\"country\"],\n", " latitude=data[\"coord\"][\"lat\"],\n", " longitude=data[\"coord\"][\"lon\"]\n", " ),\n", " temperature=data[\"main\"][\"temp\"],\n", " conditions=data[\"weather\"][0][\"description\"],\n", " humidity=data[\"main\"][\"humidity\"],\n", " wind_speed=data[\"wind\"][\"speed\"]\n", " )\n", " \n", " return weather.dict()\n", " \n", " except Exception as e:\n", " return {\"error\": str(e)}\n", "\n", "# For demo, we'll simulate the API response\n", "async def mock_weather_api():\n", " return {\n", " \"name\": \"San Francisco\",\n", " \"sys\": {\"country\": \"US\"},\n", " \"coord\": {\"lat\": 37.7749, \"lon\": -122.4194},\n", " \"main\": {\"temp\": 18.5, \"humidity\": 65},\n", " \"weather\": [{\"description\": \"partly cloudy\"}],\n", " \"wind\": {\"speed\": 4.2}\n", " }\n", "\n", "# Test our resource\n", "with httpx.MockClient() as client:\n", " client.get = mock_weather_api # Use our mock\n", " result = asyncio.run(get_weather_data(\"San Francisco\"))\n", " print(\"🌤️ Weather Data:\", json.dumps(result, indent=2))\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Now let's implement our API tools\n", "@mcp.tool()\n", "async def get_forecast(location: str, days: int = 5) -> Dict:\n", " \"\"\"\n", " Get weather forecast for a location\n", " \n", " Args:\n", " location: City name (e.g., 'San Francisco')\n", " days: Number of forecast days (1-5)\n", " \n", " Returns:\n", " Dictionary with forecast data\n", " \"\"\"\n", " try:\n", " # First get current weather from resource\n", " weather = await get_weather_data(location)\n", " \n", " if \"error\" in weather:\n", " raise Exception(weather[\"error\"])\n", " \n", " # Make forecast API request\n", " async with httpx.AsyncClient() as client:\n", " response = await client.get(\n", " f\"{config.base_url}/forecast\",\n", " params={\n", " \"q\": location,\n", " \"appid\": config.api_key,\n", " \"units\": \"metric\",\n", " \"cnt\": days\n", " },\n", " timeout=config.timeout\n", " )\n", " \n", " if response.status_code == 429:\n", " raise Exception(\"Rate limit exceeded\")\n", " \n", " response.raise_for_status()\n", " data = response.json()\n", " \n", " # Parse forecast data\n", " forecast_days = []\n", " for day in data[\"list\"][:days]:\n", " forecast_days.append(\n", " ForecastDay(\n", " date=datetime.fromtimestamp(day[\"dt\"]),\n", " temp_high=day[\"main\"][\"temp_max\"],\n", " temp_low=day[\"main\"][\"temp_min\"],\n", " conditions=day[\"weather\"][0][\"description\"],\n", " precipitation=day[\"rain\"][\"3h\"] if \"rain\" in day else 0.0\n", " )\n", " )\n", " \n", " forecast = Forecast(\n", " location=Location(**weather[\"location\"]),\n", " days=forecast_days\n", " )\n", " \n", " return {\n", " \"success\": True,\n", " \"data\": forecast.dict()\n", " }\n", " \n", " except Exception as e:\n", " return {\n", " \"success\": False,\n", " \"error\": str(e),\n", " \"prompt\": \"error_prompt\"\n", " }\n", "\n", "# For demo, we'll simulate the forecast API\n", "async def mock_forecast_api():\n", " return {\n", " \"list\": [\n", " {\n", " \"dt\": datetime.now().timestamp(),\n", " \"main\": {\"temp_max\": 20.5, \"temp_min\": 15.2},\n", " \"weather\": [{\"description\": \"sunny\"}],\n", " \"rain\": {\"3h\": 0.0}\n", " },\n", " {\n", " \"dt\": (datetime.now().timestamp() + 86400),\n", " \"main\": {\"temp_max\": 22.0, \"temp_min\": 16.5},\n", " \"weather\": [{\"description\": \"cloudy\"}],\n", " \"rain\": {\"3h\": 2.5}\n", " }\n", " ]\n", " }\n", "\n", "# Test our forecast tool\n", "with httpx.MockClient() as client:\n", " client.get = mock_forecast_api # Use our mock\n", " result = asyncio.run(get_forecast(\"San Francisco\", days=2))\n", " print(\"🌡️ Forecast:\", json.dumps(result, indent=2))\n" ] }, { "cell_type": "raw", "metadata": { "vscode": { "languageId": "raw" } }, "source": [ "# 🚀 Running the FastMCP API Server\n", "\n", "Now that we have our API integration server built with FastMCP patterns, let's run it:\n", "\n", "```python\n", "if __name__ == \"__main__\":\n", " # Run the MCP server\n", " print(\"🌐 Starting API Integration MCP Server...\")\n", " mcp.run(transport=\"streamable-http\")\n", "```\n", "\n", "## 🎯 Key Takeaways\n", "\n", "1. **Type Safety**\n", " - Use Pydantic models for API data\n", " - Validate requests and responses\n", " - Handle API-specific types\n", "\n", "2. **Resource Providers**\n", " - Cache API responses\n", " - Handle rate limiting\n", " - Provide consistent interfaces\n", "\n", "3. **System Prompts**\n", " - Guide API interactions\n", " - Handle API errors gracefully\n", " - Provide clear error messages\n", "\n", "4. **Best Practices**\n", " - Use async HTTP clients\n", " - Implement proper error handling\n", " - Follow API rate limits\n", " - Structure responses consistently\n", "\n", "## 🔜 Next Steps\n", "\n", "Continue to [Database Operations](08_database_operations.ipynb) to learn how to integrate your MCP server with databases!\n" ] }, { "cell_type": "raw", "metadata": { "vscode": { "languageId": "raw" } }, "source": [ "# 🌐 API Integration with MCP\n", "\n", "Master the art of connecting MCP servers to external APIs! This notebook covers authentication, rate limiting, caching, and building robust API integrations that AI assistants can rely on.\n", "\n", "## 🎯 Learning Objectives\n", "\n", "By the end of this notebook, you will:\n", "- Integrate REST APIs with MCP servers\n", "- Handle authentication (API keys, OAuth, JWT)\n", "- Implement rate limiting and error retry logic\n", "- Build caching strategies for performance\n", "- Create API abstraction layers\n", "\n", "## 🛠️ What You'll Build\n", "\n", "1. **🔑 Authenticated API Client** - Secure API integration patterns\n", "2. **📊 Social Media MCP** - Twitter/X API integration\n", "3. **💰 Payment Gateway MCP** - Stripe payment processing\n", "4. **📧 Email Service MCP** - SendGrid email automation\n", "5. **🌍 Maps & Geocoding MCP** - Google Maps integration\n", "\n", "## 📚 Table of Contents\n", "\n", "1. [API Authentication Patterns](#authentication)\n", "2. [REST API Integration](#rest-integration)\n", "3. [Rate Limiting & Retries](#rate-limiting)\n", "4. [Caching Strategies](#caching)\n", "5. [Real-world Examples](#examples)\n", "\n", "---\n" ] } ], "metadata": { "language_info": { "name": "python" } }, "nbformat": 4, "nbformat_minor": 2 }