Budowa serwera MCP w Pythonie — architektura, FastMCP i pułapki wdrożenia

Na tym blogu korzystam z własnego serwera MCP do pracy z treściami WordPressa. To praktyczny punkt wyjścia do tego artykułu: architektury protokołu, decyzji implementacyjnych i szkieletu serwera w Pythonie. Zakres zmian, które serwer może wykonać, zależy od faktycznie udostępnionych narzędzi i uprawnień.

Model Context Protocol (MCP) to otwarty standard integracji aplikacji AI z narzędziami i danymi. Stan na 27 września 2026: aktualna specyfikacja ma numer 2026-07-28, a oficjalny Python SDK jest w linii 2.x. Przykład FastMCP poniżej zachowuje API linii 1.x — wymagania wersji podaję obok kodu, żeby nie mieszać dwóch generacji interfejsu.

Co protokół MCP naprawdę rozwiązuje

Problem, który MCP adresuje, jest kombinatoryczny. Masz M aplikacji LLM (Claude Desktop, Cursor, VS Code, ChatGPT) i N zewnętrznych systemów (baza danych, GitHub, API firmowe, WordPress). Bez wspólnego standardu każda para wymaga dedykowanej integracji — to M×N implementacji, każda z własnym formatem, autoryzacją i cyklem utrzymania. MCP zamienia to na M+N: piszesz serwer raz, a każdy zgodny klient potrafi go odkryć i użyć bez linijki kodu po swojej stronie.

Mechanicznie MCP stoi na JSON-RPC 2.0 i definiuje trzy role. Host to aplikacja LLM, która koordynuje całość. Klient jest instancjonowany przez hosta — jeden klient na jeden serwer. Serwer dostarcza kontekst i możliwości. To celowa inspiracja Language Server Protocolem: tak jak LSP ustandaryzował wsparcie języków w edytorach, MCP standaryzuje wpinanie narzędzi i danych do ekosystemu AI.

Function calling i MCP rozwiązują problemy na różnych poziomach. Pierwsze pozwala modelowi proponować wywołania funkcji; MCP opisuje, jak klient odkrywa i wywołuje możliwości zewnętrznego serwera. W rewizji 2026-07-28 wersja protokołu jest deklarowana w każdym żądaniu. Starsze rewizje używały uzgadniania przy inicjalizacji.

Trzy prymitywy: tools, resources i prompts

Serwer MCP wystawia możliwości przez trzy prymitywy. Mylenie ich to najczęstszy błąd projektowy — każdy ma inny kontrakt i inne zastosowanie.

Reguła praktyczna: Tool gdy potrzebujesz walidacji wejścia i logiki biznesowej („utwórz wpis o tytule X i statusie Y”). Resource gdy udostępniasz dane pod prostym parametrem („treść dokumentu o nazwie Z”). Prompt gdy dajesz użytkownikowi gotowy, sparametryzowany scenariusz. W praktyce większość serwerów zaczyna i kończy na tools — reszta to optymalizacja kontekstu.

Transport: stdio vs Streamable HTTP

MCP definiuje dwa transporty i wybór między nimi to pierwsza architektoniczna decyzja przy budowie serwera MCP.

Rewizja 2026-07-28 przenosi metadane protokołu do poszczególnych żądań i udostępnia server/discover. Bezsesyjny model ułatwia kierowanie żądań do różnych instancji. Wdrożenie nadal musi jednak rozwiązać własny stan aplikacji, autoryzację i zgodność ze starszymi klientami; sam load balancer tego nie załatwia.

To nie znaczy, że Twoja aplikacja musi być bezstanowa. Serwer, który potrzebuje stanu między wywołaniami, robi to, co API HTTP robiły od zawsze: wystawia jawny uchwyt (np. basket_id) z jednego narzędzia i każe modelowi podawać go z powrotem jako zwykły argument w kolejnych wywołaniach. Projektuj więc pod bezstanowy transport od początku — to kierunek, w którym idzie protokół, i tańsza droga do skalowania.

Szkielet serwera FastMCP — przykład dla SDK 1.x

Ten przykład korzysta z mcp.server.fastmcp.FastMCP z utrzymywanej linii SDK 1.x. Przypnij zależność mcp>=1.28,<2, a także wersje httpx i Pydantic 2 w pliku zależności projektu. SDK 2.x ma zmienione API — migrację trzeba przeprowadzić osobno. Kod pokazuje walidację wejścia i asynchroniczne I/O, ale nie jest kompletnym wdrożeniem produkcyjnym.

API_BASE oraz pola odpowiedzi pogodowej są umowne. Zastąp je kontraktem rzeczywistego dostawcy i dodaj walidację jego odpowiedzi. Bez tego przykład nie pobierze prawdziwej prognozy.

from __future__ import annotations

import os
import httpx
from pydantic import BaseModel, Field, ConfigDict
from mcp.server.fastmcp import FastMCP

# Serwer nazywamy wg konwencji {usługa}_mcp
mcp = FastMCP("weather_mcp", port=8000)

API_BASE = "https://api.example-weather.com/v1"


class ForecastInput(BaseModel):
    """Walidacja wejścia dla zapytania o prognozę."""
    model_config = ConfigDict(
        str_strip_whitespace=True,
        extra="forbid",          # odrzuć nieznane pola
    )

    city: str = Field(..., description="Nazwa miasta, np. 'Wrocław'",
                      min_length=1, max_length=100)
    days: int = Field(default=3, description="Horyzont prognozy w dniach",
                      ge=1, le=14)


def _handle_error(e: Exception) -> str:
    """Spójne, pomocne komunikaty błędów dla modelu."""
    if isinstance(e, httpx.HTTPStatusError):
        code = e.response.status_code
        if code == 404:
            return "Błąd: nie znaleziono miasta. Sprawdź pisownię nazwy."
        if code == 429:
            return "Błąd: przekroczono limit zapytań. Odczekaj chwilę."
        return f"Błąd: API zwróciło status {code}."
    if isinstance(e, httpx.TimeoutException):
        return "Błąd: przekroczono czas oczekiwania. Spróbuj ponownie."
    return f"Błąd: nieoczekiwany wyjątek: {type(e).__name__}"


@mcp.tool(
    name="get_forecast",
    annotations={
        "title": "Pobierz prognozę pogody",
        "readOnlyHint": True,      # nie modyfikuje stanu
        "openWorldHint": True,     # sięga do zewnętrznego API
    },
)
async def get_forecast(params: ForecastInput) -> str:
    """Zwraca prognozę pogody dla miasta.

    Args:
        params: zwalidowane wejście (city, days).
    Returns:
        str: sformatowana prognoza albo pomocny komunikat błędu.
    """
    api_key = os.environ.get("WEATHER_API_KEY")
    if not api_key:
        return "Błąd konfiguracji: brak WEATHER_API_KEY w środowisku."

    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            resp = await client.get(
                f"{API_BASE}/forecast",
                params={"q": params.city, "days": params.days},
                headers={"Authorization": f"Bearer {api_key}"},
            )
            resp.raise_for_status()
            data = resp.json()
    except Exception as e:
        return _handle_error(e)

    lines = [f"Prognoza dla {params.city} ({params.days} dni):"]
    for day in data["forecast"]:
        lines.append(f"  {day['date']}: {day['temp_c']}°C, {day['condition']}")
    return "\n".join(lines)


if __name__ == "__main__":
    mcp.run()   # transport stdio (domyślny)

Kilka rzeczy jest tu nieprzypadkowych. Model Pydantic z extra="forbid" odrzuca nieznane pola, zamiast je po cichu ignorować. Adnotacje w dekoratorze (readOnlyHint, openWorldHint) to sygnały dla hosta. Całe I/O jest asynchroniczne. A sekret idzie ze zmiennej środowiskowej, nie z kodu — do tego wrócę przy bezpieczeństwie.

Obsługa błędów, która pomaga modelowi

Zwróć uwagę na funkcję _handle_error powyżej. To nie jest kosmetyka. Komunikat błędu w serwerze MCP jest czytany przez model, nie przez człowieka wpatrzonego w logi — i decyduje, czy model spróbuje sensownie naprawić wywołanie, czy utknie. „Błąd 404″ nie mówi nic; „nie znaleziono miasta, sprawdź pisownię” mówi modelowi, co zrobić dalej. Traktuj każdy komunikat jak instrukcję naprawczą, nie jak wpis do logu.

To ta sama dyscyplina, co przy debugowaniu jako procesie dedukcji, nie zgadywania — precyzyjny sygnał zamiast szumu skraca drogę do przyczyny. Różnica polega na tym, że tutaj odbiorcą sygnału jest model, który na jego podstawie planuje kolejny krok.

Bezpieczeństwo: dlaczego opisy narzędzi są niezaufane

Specyfikacja MCP mówi wprost: narzędzia to dowolne wykonanie kodu i należy je traktować z odpowiednią ostrożnością. Co więcej — opisy zachowania narzędzia, w tym adnotacje, są niezaufane, chyba że pochodzą z zaufanego serwera. To nie formalność. Złośliwy serwer może w opisie narzędzia albo w wyniku jego wywołania przemycić instrukcje, które model potraktuje jak polecenie — to prompt injection przez wynik narzędzia.

Sekrety pobieraj z konfiguracji środowiska lub menedżera sekretów, a nie z kodu czy opisów narzędzi. Dla chronionego zdalnego serwera wdroż autoryzację zgodną z używaną rewizją MCP, sprawdzaj tokeny i uprawnienia po stronie serwera. Pydantic waliduje kształt danych, ale nie rozstrzyga, czy dany użytkownik może wykonać operację. Adnotacje opisuj zgodnie z zachowaniem narzędzia:

Host może korzystać z adnotacji przy prezentowaniu narzędzia i proszeniu o zgodę. Są jednak wskazówkami, nie zabezpieczeniem: readOnlyHint nie blokuje zapisu, a destructiveHint nie zastępuje kontroli uprawnień.

Stan, współbieżność i skalowanie

Serwer produkcyjny obsługuje wielu klientów naraz, a każde narzędzie robi I/O — zapytanie do API, do bazy, do dysku. Dlatego cały kod jest asynchroniczny (async def, httpx.AsyncClient): jeden proces obsługuje wiele równoległych wywołań bez blokowania, bo w czasie oczekiwania na odpowiedź sieciową event loop przełącza się na inne zadanie.

Oczekiwanie na sieć i pracę pętli zdarzeń rozwijam w porównaniu epoll i io_uring. async def pomaga współdzielić czas oczekiwania, ale nie usuwa blokującego kodu ani kosztownych obliczeń. Skalowanie serwera wymaga też limitów współbieżności, timeoutów i przemyślanego zarządzania połączeniami.

# Wybierz jeden transport przy uruchomieniu serwera z przykładu.
# Lokalnie, przez standardowe wejście i wyjście:
mcp.run()

# Alternatywnie: HTTP w SDK 1.x; port ustawiono w konstruktorze.
# mcp.run(transport="streamable-http")

Podsumowanie

Najpierw ustal, jakie operacje serwer udostępnia i kto może je wywołać. Potem dobierz transport, walidację, obsługę błędów i sposób przechowywania stanu. Framework oszczędza kod protokołu; odpowiedzialność za skutki narzędzi zostaje po Twojej stronie.

Przy aktualizacji sprawdzaj osobno wersję protokołu, SDK i możliwości klienta. Jawny identyfikator stanu aplikacji ułatwia skalowanie, ale sam w sobie nie jest dowodem zgodności z nową specyfikacją. To pierwszy wpis z serii o MCP — w kolejnych wejdziemy głębiej w bezpieczeństwo i wzorce zaawansowane.

Weryfikacja techniczna: wersjonowanie MCP, transporty 2026-07-28, Python SDK 1.x, Python SDK 2.x i migracja.