fastapi and fastmcp

FastAPI to najszybciej rozwijajacy sie framework webowy Pythona — idealny do budowy API i backendu dla aplikacji AI. FastMCP to biblioteka do tworzenia serwerow MCP (Model Context Protocol), ktore pozwalaja agentom AI laczyc sie z zewnetrznymi narzedziami. Ten artykul pokazuje jak uzywac obu technologii do budowy nowoczesnych aplikacji AI.

Czesc I: FastAPI

FastAPI to nowoczesny framework do budowy API w Pythonie. Laczy szybkosc (porownywalna z Node.js i Go) z produktywnoscia Pythona i automatyczna dokumentacja.

Instalacja

# Podstawowa instalacja
pip install fastapi

# Z serwerem ASGI (wymagane do uruchomienia)
pip install fastapi[standard]
# lub osobno:
pip install uvicorn

# Wszystkie dodatki
pip install fastapi[all]

Hello World

# main.py
from fastapi import FastAPI

app = FastAPI(
    title="My AI API",
    description="API dla aplikacji AI",
    version="1.0.0"
)

@app.get("/")
def root():
    return {"message": "Hello World"}

@app.get("/health")
def health_check():
    return {"status": "healthy"}

# Uruchomienie:
# uvicorn main:app --reload
# 
# Dokumentacja automatyczna:
# http://localhost:8000/docs     (Swagger UI)
# http://localhost:8000/redoc    (ReDoc)

Path i Query Parameters

from fastapi import FastAPI, Query, Path
from typing import Optional

app = FastAPI()

# Path parameters
@app.get("/users/{user_id}")
def get_user(
    user_id: int = Path(..., title="User ID", ge=1)
):
    return {"user_id": user_id}

# Query parameters
@app.get("/items/")
def list_items(
    skip: int = Query(default=0, ge=0),
    limit: int = Query(default=10, le=100),
    search: Optional[str] = Query(default=None, min_length=1)
):
    return {
        "skip": skip,
        "limit": limit,
        "search": search
    }

# GET /items/?skip=0&limit=10&search=laptop

Request Body z Pydantic

from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional
from enum import Enum

app = FastAPI()

# Enum dla walidacji
class Priority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

# Request model
class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=100)
    description: Optional[str] = Field(default=None, max_length=1000)
    priority: Priority = Priority.MEDIUM
    
    model_config = {
        "json_schema_extra": {
            "examples": [{
                "title": "Implement feature",
                "description": "Add new AI capability",
                "priority": "high"
            }]
        }
    }

# Response model
class TaskResponse(BaseModel):
    id: int
    title: str
    description: Optional[str]
    priority: Priority
    completed: bool = False

# Endpoint z walidacja wejscia i wyjscia
@app.post("/tasks/", response_model=TaskResponse)
def create_task(task: TaskCreate):
    # task jest juz zwalidowany przez Pydantic
    return TaskResponse(
        id=1,
        title=task.title,
        description=task.description,
        priority=task.priority
    )

Async Endpoints

from fastapi import FastAPI
import httpx
import asyncio

app = FastAPI()

# Async endpoint - lepszy dla I/O operations
@app.get("/external-data")
async def fetch_external_data():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

# Rownolegle wywolania
@app.get("/parallel")
async def parallel_calls():
    async with httpx.AsyncClient() as client:
        # Wykonaj rownolegle
        results = await asyncio.gather(
            client.get("https://api1.example.com"),
            client.get("https://api2.example.com"),
            client.get("https://api3.example.com")
        )
        return [r.json() for r in results]

# Sync jest OK dla CPU-bound lub prostych operacji
@app.get("/sync")
def sync_endpoint():
    return {"status": "ok"}

Dependency Injection

from fastapi import FastAPI, Depends, HTTPException, Header
from typing import Annotated

app = FastAPI()

# Prosta zaleznosc
def get_db():
    """Symulacja polaczenia z baza."""
    db = {"connection": "active"}
    try:
        yield db
    finally:
        # Cleanup
        db["connection"] = "closed"

# Zaleznosc z walidacja
async def verify_api_key(
    x_api_key: Annotated[str, Header()]
):
    if x_api_key != "secret-key":
        raise HTTPException(status_code=401, detail="Invalid API key")
    return x_api_key

# Zaleznosc z parametrami
def pagination(
    skip: int = 0,
    limit: int = 10
):
    return {"skip": skip, "limit": limit}

# Uzycie zaleznosci
@app.get("/items/")
async def list_items(
    db: Annotated[dict, Depends(get_db)],
    api_key: Annotated[str, Depends(verify_api_key)],
    pagination: Annotated[dict, Depends(pagination)]
):
    return {
        "db_status": db["connection"],
        "pagination": pagination
    }

# Globalna zaleznosc dla wszystkich endpointow
app = FastAPI(dependencies=[Depends(verify_api_key)])

API dla aplikacji AI

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from openai import AsyncOpenAI
import asyncio

app = FastAPI(title="AI Chat API")
client = AsyncOpenAI()

class ChatRequest(BaseModel):
    message: str
    model: str = "gpt-4o"
    temperature: float = 0.7

class ChatResponse(BaseModel):
    response: str
    model: str
    tokens_used: int

# Standardowe wywolanie
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    try:
        response = await client.chat.completions.create(
            model=request.model,
            messages=[{"role": "user", "content": request.message}],
            temperature=request.temperature
        )
        return ChatResponse(
            response=response.choices[0].message.content,
            model=response.model,
            tokens_used=response.usage.total_tokens
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# Streaming response
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate():
        stream = await client.chat.completions.create(
            model=request.model,
            messages=[{"role": "user", "content": request.message}],
            temperature=request.temperature,
            stream=True
        )
        async for chunk in stream:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content
    
    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )

Error Handling i Middleware

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
import time
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # W produkcji: konkretne domeny
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Custom middleware - logging czasu
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    duration = time.time() - start_time
    
    logger.info(
        f"{request.method} {request.url.path} "
        f"- {response.status_code} - {duration:.3f}s"
    )
    
    response.headers["X-Process-Time"] = str(duration)
    return response

# Custom exception handler
class AIServiceError(Exception):
    def __init__(self, message: str, code: str):
        self.message = message
        self.code = code

@app.exception_handler(AIServiceError)
async def ai_error_handler(request: Request, exc: AIServiceError):
    return JSONResponse(
        status_code=500,
        content={
            "error": exc.code,
            "message": exc.message,
            "path": str(request.url)
        }
    )

Czesc II: FastMCP

FastMCP to biblioteka do tworzenia serwerow MCP (Model Context Protocol) — otwartego standardu Anthropic umozliwiajacego agentom AI laczenie sie z zewnetrznymi narzedziami i zrodlami danych.

Instalacja FastMCP

# Instalacja
pip install fastmcp

# Lub z poetry
poetry add fastmcp

# Sprawdz wersje
pip show fastmcp

Pierwszy serwer MCP

# server.py
from fastmcp import FastMCP

# Utworz serwer MCP
mcp = FastMCP(
    name="My Tools Server",
    version="1.0.0"
)

# Proste narzedzie
@mcp.tool()
def greet(name: str) -> str:
    """Pozdrow uzytkownika po imieniu.
    
    Args:
        name: Imie uzytkownika do pozdrowienia
    """
    return f"Czesc, {name}! Milo Cie poznac."

# Uruchomienie serwera
if __name__ == "__main__":
    mcp.run()

# Uruchom:
# python server.py
# lub:
# fastmcp run server.py

Narzedzia (Tools)

Narzedzia to funkcje ktore agent AI moze wywolywac:

from fastmcp import FastMCP
from pydantic import Field
from typing import Annotated
import httpx

mcp = FastMCP("AI Tools")

# Narzedzie z walidacja Pydantic
@mcp.tool()
def calculate(
    expression: Annotated[str, Field(description="Wyrazenie matematyczne")]
) -> str:
    """Oblicza wyrazenie matematyczne.
    
    Args:
        expression: Wyrazenie do obliczenia, np. "2 + 2", "10 * 5"
    """
    try:
        result = eval(expression)
        return f"Wynik: {result}"
    except Exception as e:
        return f"Blad: {str(e)}"

# Narzedzie async z zewnetrznym API
@mcp.tool()
async def get_weather(city: str) -> str:
    """Pobiera aktualna pogode dla miasta.
    
    Args:
        city: Nazwa miasta (np. Warszawa, Krakow)
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://wttr.in/{city}?format=3"
        )
        return response.text

# Narzedzie z wieloma parametrami
@mcp.tool()
def search_database(
    query: Annotated[str, Field(description="Fraza wyszukiwania")],
    limit: Annotated[int, Field(default=10, ge=1, le=100)] = 10,
    category: Annotated[str, Field(default="all")] = "all"
) -> str:
    """Przeszukuje baze danych produktow.
    
    Args:
        query: Fraza wyszukiwania
        limit: Maksymalna liczba wynikow (1-100)
        category: Kategoria produktow (all, electronics, books, etc.)
    """
    # Symulacja wyszukiwania
    results = [
        {"id": i, "name": f"Product {i}", "category": category}
        for i in range(limit)
    ]
    return str(results)

Zasoby (Resources)

Zasoby to dane ktore agent moze odczytywac:

from fastmcp import FastMCP
import json
from pathlib import Path

mcp = FastMCP("Data Server")

# Statyczny zasob
@mcp.resource("config://app")
def get_app_config() -> str:
    """Konfiguracja aplikacji."""
    config = {
        "version": "1.0.0",
        "environment": "production",
        "features": ["ai", "search", "export"]
    }
    return json.dumps(config, indent=2)

# Dynamiczny zasob z parametrem
@mcp.resource("file://{path}")
def read_file(path: str) -> str:
    """Odczytuje zawartosc pliku.
    
    Args:
        path: Sciezka do pliku
    """
    file_path = Path(path)
    if not file_path.exists():
        return f"Plik nie istnieje: {path}"
    return file_path.read_text()

# Zasob z bazy danych
@mcp.resource("db://users/{user_id}")
async def get_user(user_id: str) -> str:
    """Pobiera dane uzytkownika z bazy.
    
    Args:
        user_id: ID uzytkownika
    """
    # Symulacja bazy danych
    user = {
        "id": user_id,
        "name": "Jan Kowalski",
        "email": "jan@example.com"
    }
    return json.dumps(user)

Prompty (Prompts)

Prompty to szablony promptow ktore agent moze uzywac:

from fastmcp import FastMCP
from fastmcp.prompts import Prompt

mcp = FastMCP("Prompt Library")

# Prosty prompt
@mcp.prompt()
def code_review(code: str, language: str = "python") -> str:
    """Prompt do code review.
    
    Args:
        code: Kod do przegladniecia
        language: Jezyk programowania
    """
    return f"""Przejrzyj ponizszy kod {language} i podaj:
1. Potencjalne bledy
2. Sugestie optymalizacji
3. Zgodnosc z best practices

Kod:
```{language}
{code}
```"""

# Prompt z wieloma wiadomosciami
@mcp.prompt()
def analyze_data(data: str, analysis_type: str) -> list[dict]:
    """Prompt do analizy danych.
    
    Args:
        data: Dane do analizy (JSON lub CSV)
        analysis_type: Typ analizy (summary, trends, anomalies)
    """
    return [
        {
            "role": "system",
            "content": f"Jestes ekspertem od analizy danych. Wykonaj analize typu: {analysis_type}"
        },
        {
            "role": "user",
            "content": f"Przeanalizuj te dane:\n\n{data}"
        }
    ]

Kontekst i Lifecycle

from fastmcp import FastMCP, Context
from contextlib import asynccontextmanager
import asyncpg

# Lifecycle - inicjalizacja i cleanup
@asynccontextmanager
async def lifespan(mcp: FastMCP):
    """Lifecycle serwera - setup i teardown."""
    # Startup
    pool = await asyncpg.create_pool("postgresql://localhost/mydb")
    mcp.state["db_pool"] = pool
    print("Database pool created")
    
    yield
    
    # Shutdown
    await pool.close()
    print("Database pool closed")

mcp = FastMCP("DB Server", lifespan=lifespan)

# Uzycie kontekstu w narzedziu
@mcp.tool()
async def query_database(sql: str, ctx: Context) -> str:
    """Wykonuje zapytanie SQL.
    
    Args:
        sql: Zapytanie SQL (tylko SELECT)
    """
    if not sql.strip().upper().startswith("SELECT"):
        return "Tylko zapytania SELECT sa dozwolone"
    
    pool = ctx.state["db_pool"]
    async with pool.acquire() as conn:
        rows = await conn.fetch(sql)
        return str([dict(row) for row in rows])

Kompletny serwer MCP

# complete_server.py
from fastmcp import FastMCP, Context
from pydantic import Field
from typing import Annotated
import httpx
import json
from datetime import datetime

mcp = FastMCP(
    name="AI Assistant Tools",
    version="1.0.0",
    description="Narzedzia dla asystenta AI"
)

# ========== TOOLS ==========

@mcp.tool()
def get_current_time() -> str:
    """Zwraca aktualny czas i date."""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@mcp.tool()
async def web_search(
    query: Annotated[str, Field(description="Fraza wyszukiwania")],
    num_results: Annotated[int, Field(default=5, ge=1, le=10)] = 5
) -> str:
    """Wyszukuje informacje w internecie.
    
    Args:
        query: Fraza wyszukiwania
        num_results: Liczba wynikow (1-10)
    """
    # Symulacja wyszukiwania
    results = [
        {"title": f"Result {i}", "url": f"https://example.com/{i}"}
        for i in range(num_results)
    ]
    return json.dumps(results, indent=2)

@mcp.tool()
def create_note(
    title: Annotated[str, Field(min_length=1, max_length=100)],
    content: Annotated[str, Field(min_length=1)],
    tags: Annotated[list[str], Field(default=[])] = []
) -> str:
    """Tworzy nowa notatke.
    
    Args:
        title: Tytul notatki
        content: Tresc notatki
        tags: Lista tagow
    """
    note = {
        "id": "note-123",
        "title": title,
        "content": content,
        "tags": tags,
        "created_at": datetime.now().isoformat()
    }
    return json.dumps(note, indent=2)

# ========== RESOURCES ==========

@mcp.resource("notes://list")
def list_notes() -> str:
    """Lista wszystkich notatek."""
    notes = [
        {"id": "1", "title": "Meeting notes"},
        {"id": "2", "title": "Project ideas"}
    ]
    return json.dumps(notes)

@mcp.resource("notes://{note_id}")
def get_note(note_id: str) -> str:
    """Pobiera notatke po ID."""
    return json.dumps({
        "id": note_id,
        "title": "Sample Note",
        "content": "This is the content..."
    })

# ========== PROMPTS ==========

@mcp.prompt()
def summarize(text: str, style: str = "concise") -> str:
    """Prompt do podsumowania tekstu."""
    return f"""Podsumuj ponizszy tekst w stylu {style}:

{text}

Podaj kluczowe punkty i wnioski."""

# Uruchomienie
if __name__ == "__main__":
    mcp.run()

Konfiguracja klienta MCP

// Claude Desktop: ~/Library/Application Support/Claude/claude_desktop_config.json
// lub Windows: %APPDATA%/Claude/claude_desktop_config.json

{
  "mcpServers": {
    "my-tools": {
      "command": "python",
      "args": ["/path/to/complete_server.py"],
      "env": {
        "PYTHONPATH": "/path/to/project"
      }
    },
    "another-server": {
      "command": "uvx",
      "args": ["--from", "fastmcp", "run", "server.py"]
    }
  }
}

Czesc III: FastAPI + FastMCP

Mozesz laczyc FastAPI i FastMCP w jednej aplikacji:

# combined_server.py
from fastapi import FastAPI
from fastmcp import FastMCP
from pydantic import BaseModel

# FastAPI app
app = FastAPI(title="AI Service")

# FastMCP server
mcp = FastMCP("AI Tools")

# Wspoldzielona logika
class NoteService:
    notes = {}
    
    @classmethod
    def create(cls, title: str, content: str) -> dict:
        note_id = str(len(cls.notes) + 1)
        note = {"id": note_id, "title": title, "content": content}
        cls.notes[note_id] = note
        return note
    
    @classmethod
    def get(cls, note_id: str) -> dict | None:
        return cls.notes.get(note_id)
    
    @classmethod
    def list_all(cls) -> list:
        return list(cls.notes.values())

# ========== FastAPI endpoints ==========

class NoteCreate(BaseModel):
    title: str
    content: str

class NoteResponse(BaseModel):
    id: str
    title: str
    content: str

@app.post("/api/notes", response_model=NoteResponse)
def create_note_api(note: NoteCreate):
    return NoteService.create(note.title, note.content)

@app.get("/api/notes", response_model=list[NoteResponse])
def list_notes_api():
    return NoteService.list_all()

@app.get("/api/notes/{note_id}", response_model=NoteResponse)
def get_note_api(note_id: str):
    note = NoteService.get(note_id)
    if not note:
        raise HTTPException(404, "Note not found")
    return note

# ========== MCP tools (ta sama logika!) ==========

@mcp.tool()
def create_note(title: str, content: str) -> str:
    """Tworzy nowa notatke."""
    note = NoteService.create(title, content)
    return f"Utworzono notatke: {note}"

@mcp.tool()
def list_notes() -> str:
    """Lista wszystkich notatek."""
    return str(NoteService.list_all())

@mcp.tool()
def get_note(note_id: str) -> str:
    """Pobiera notatke po ID."""
    note = NoteService.get(note_id)
    return str(note) if note else "Note not found"

# Uruchomienie obu serwerow
if __name__ == "__main__":
    import threading
    import uvicorn
    
    # FastAPI w osobnym urzadzeniu
    api_thread = threading.Thread(
        target=lambda: uvicorn.run(app, host="0.0.0.0", port=8000)
    )
    api_thread.start()
    
    # MCP server
    mcp.run()

Kiedy uzywac czego?

Podsumowanie API

Funkcja	FastAPI	FastMCP
Inicjalizacja	app = FastAPI()	mcp = FastMCP()
Endpoint/Tool	@app.get(“/path”)	@mcp.tool()
Dane statyczne	StaticFiles	@mcp.resource()
Szablony	Jinja2Templates	@mcp.prompt()
Walidacja	Pydantic models	Pydantic + Field
Uruchomienie	uvicorn main:app	mcp.run()
Dokumentacja	/docs (Swagger)	Automatyczna (MCP)