Serwer MCP w Pythonie z FastMCP: narzędzia, zasoby i prompty

Narzędzia agenta z poprzednich części żyły w tym samym programie co agent. Działa to, dopóki jest jeden program. Gdy to samo narzędzie ma być dostępne dla kilku aplikacji (asystenta w edytorze, czatu, własnego agenta), kopiowanie kodu przestaje mieć sens. Rozwiązaniem jest MCP, czyli Model Context Protocol: standardowy sposób, w jaki aplikacje AI podłączają zewnętrzne narzędzia i dane. Wynosimy narzędzia do osobnego serwera, a każdy klient MCP może z niego korzystać.

To piąta część cyklu. Serwer opakuje te same narzędzia, które zbudowaliśmy w agencie z wyszukiwaniem i odczytem dokumentów, więc potrzebujesz jego kodu (docs_tools.py) i indeksu z wyszukiwarki semantycznej. W kolejnej części napiszemy klienta MCP i podłączymy go do agenta (agent loop) z tool callingiem w OpenAI API.

Czym jest MCP i co serwer może udostępniać?

Serwer MCP udostępnia klientom trzy rodzaje rzeczy. Różnią się tym, kto decyduje o ich użyciu:

Z poprzednich artykułów znasz tylko pierwszy z nich. Tool calling z OpenAI API to rozmowa między Twoim kodem a modelem. MCP to osobna warstwa między aplikacją a narzędziami: klient pyta serwer, jakie ma narzędzia, i dopiero te definicje przekazuje modelowi. Dzięki temu narzędzia opisujesz raz, w serwerze, a nie w każdej aplikacji osobno.

Jak zainstalować SDK i dlaczego przypinamy wersję?

Instalujemy SDK w wersji 1.x:

pip install "mcp<2"

Używamy mcp<2, bo wersja 2 wprowadziła zmiany niezgodne wstecz: klasę FastMCP przemianowano na MCPServer i zmieniono jej import, więc kod z tego cyklu działa z wersją 1.x (testowaliśmy na 1.30.0, wymagany Python 3.10 lub nowszy). W swoim projekcie przypnij wersję w requirements.txt, np. jako mcp>=1.30,<2.

Jak napisać pierwszy serwer MCP w FastMCP?

Serwer to zwykły plik Pythona. Opakowujemy w nim klasę DocsToolbox z poprzedniej części, czyli te same funkcje wyszukiwania i odczytu. Zapisz go jako server.py, obok docs_tools.py i katalogu index/:

import sys
from functools import lru_cache
from pathlib import Path
from typing import Annotated

from mcp.server.fastmcp import FastMCP
from pydantic import Field

from docs_tools import DocsToolbox, ListDocuments, ReadSection, SearchDocs

INDEX_DIR = Path(__file__).parent / "index"

mcp = FastMCP(
    "kursownia-docs",
    instructions="Dokumenty serwisu Kursownia: regulamin, pomoc techniczna i oferta kursów.",
)


@lru_cache(maxsize=1)
def get_box() -> DocsToolbox:
    print("Ładuję model embeddingów…", file=sys.stderr)  # stderr, nigdy stdout
    return DocsToolbox(INDEX_DIR)


@mcp.tool()
def search_docs(query: str, k: Annotated[int, Field(ge=1, le=5)] = 3) -> str:
    """Szuka w dokumentach serwisu fragmentów pasujących znaczeniem do zapytania.
    Zwraca źródło, sekcję i początek każdego fragmentu."""
    return get_box().search_docs(SearchDocs(query=query, k=k))


@mcp.tool()
def read_section(source: str, section: str) -> str:
    """Zwraca pełny tekst jednej sekcji dokumentu. Użyj po wyszukiwaniu."""
    return get_box().read_section(ReadSection(source=source, section=section))


@mcp.tool()
def list_documents() -> str:
    """Wypisuje dostępne dokumenty i ich sekcje."""
    return get_box().list_documents(ListDocuments())


if __name__ == "__main__":
    transport = sys.argv[1] if len(sys.argv) > 1 else "stdio"
    mcp.run(transport=transport)  # "stdio" albo "streamable-http"

Nie potrzebujesz tu klas Pydantic z opisami pól. FastMCP buduje schemat narzędzia z adnotacji typów (query: str, k: int), a opis z docstringa. Ograniczenie ge=1, le=5 dodajemy przez Annotated i Field, tak samo jak w modelu Pydantic. Zapytanie tools/list do naszego serwera zwróciło takie schematy:

Docstring staje się opisem narzędzia, który zobaczy model, więc pisz go dla modelu, a nie dla programisty: co robi, kiedy go użyć i co zwraca.

STDIO czy HTTP: który transport wybrać?

Ten sam serwer można uruchomić na dwa sposoby i to najczęstsze źródło pomyłek:

W trybie STDIO nie uruchamiasz serwera samodzielnie w terminalu i nie podłączasz się do niego jak do usługi. To klient startuje proces i odpowiada za jego życie. Jeśli wpiszesz python server.py w terminalu, serwer po prostu czeka na dane na standardowym wejściu i wygląda na „zawieszony”. To normalne, a nie błąd. W trybie streamable-http serwer nasłuchuje pod adresem http://127.0.0.1:8000/mcp. Starszy transport SSE nadal istnieje w SDK, ale do nowych projektów wybieraj streamable-http.

Sprawdziliśmy oba tryby klientem z SDK mcp: przez STDIO i przez HTTP dostaliśmy tę samą listę trzech narzędzi i ten sam wynik wyszukiwania ([0.60] pomoc-techniczna.md › Certyfikaty ukończenia). Serwer MCP to nie zwykłe REST API: nie wywołasz go przez Postmana pojedynczym zapytaniem GET, bo rozmowa zaczyna się od wymiany initialize, którą musi poprowadzić klient MCP.

Jak dodać zasoby i prompty?

Zasób to dane pod adresem URI. Szablon z parametrem w nawiasach klamrowych działa jak dynamiczna ścieżka:

@mcp.resource("docs://{source}")
def document(source: str) -> str:
    """Pełny tekst jednego dokumentu."""
    box = get_box()
    sekcje = box._sections()
    if source not in sekcje:
        raise ValueError(f"Nie ma dokumentu {source}. Dostępne: {', '.join(sekcje)}.")
    return "\n\n".join(
        f"## {s}\n\n" + box.read_section(ReadSection(source=source, section=s))
        for s in sekcje[source]
    )


@mcp.prompt()
def answer_from_docs(question: str) -> str:
    """Szablon: odpowiedz na pytanie wyłącznie na podstawie dokumentów."""
    return (
        "Odpowiedz na pytanie wyłącznie na podstawie dokumentów dostępnych przez narzędzia. "
        "Podaj źródło w formacie (plik › sekcja). Jeśli dokumenty nie zawierają odpowiedzi, "
        f"napisz o tym wprost.\n\nPytanie: {question}"
    )

Odczyt docs://kursy.md zwrócił nam cały dokument z sekcjami, a prompt answer_from_docs gotowy tekst polecenia z wstawionym pytaniem. Zasób, tak jak narzędzie odczytu z poprzedniej części, nie dotyka dysku: składa dokument z indeksu i sprawdza nazwę na liście znanych dokumentów. Sprawdziliśmy to na serwerze: adres docs://../../etc/passwd nie pasuje do szablonu i kończy się błędem Unknown resource, a nieistniejący dokument (docs://nieistnieje.md) komunikatem „Nie ma dokumentu” z listą dostępnych.

Które z trzech elementów implementować? Zacznij od narzędzi, bo to jedyny element, który model wywołuje samodzielnie. Zasoby i prompty dokładaj wtedy, gdy klient (na przykład edytor) ma pokazać użytkownikowi listę dokumentów albo gotowe szablony.

Dlaczego nie wolno pisać na stdout w serwerze STDIO?

W trybie STDIO standardowe wyjście serwera to kanał protokołu: tędy płyną odpowiedzi JSON. Zwykły print() wrzuca do niego tekst, który nie jest wiadomością MCP. Dlatego w get_box komunikat idzie na stderr (file=sys.stderr), a w prawdziwym serwerze użyj modułu logging, który domyślnie pisze na stderr.

Sprawdziliśmy, co się stanie przy błędzie. Dodaliśmy print("Start serwera") na początku pliku i drugi print wewnątrz narzędzia. Klient z SDK 1.30 nie zerwał połączenia, tylko zalogował komunikat Failed to parse JSONRPC message from server i poszedł dalej, a narzędzie zwróciło poprawny wynik. Nie znaczy to, że można tak robić. Inne klienty mogą być mniej tolerancyjne, a my ich nie testowaliśmy, więc zasada pozostaje: na stdout serwera STDIO piszą wyłącznie wiadomości protokołu.

Dlaczego klient zrywa połączenie przy wolnym starcie serwera?

Klient po uruchomieniu serwera wysyła initialize i czeka na odpowiedź. Klienty zwykle mają limit czasu na tę odpowiedź, więc jeśli serwer przed wysłaniem odpowiedzi ładuje duży model, połączenie może się nie udać. Zmierzyliśmy to na naszym serwerze, dwukrotnie dla każdego wariantu:

Ciężka praca nie znika, tylko przesuwa się z initialize na pierwsze wywołanie narzędzia. Dla klienta to zwykle lepsze: połączenie zostaje nawiązane od razu, a pierwsze zapytanie jest po prostu wolniejsze. Trzeci wariant w kodzie to jedna zmiana w docs_tools.py, czyli przeniesienie importu do konstruktora:

class DocsToolbox:
    def __init__(self, index_dir: Path = Path("index")):
        from sentence_transformers import SentenceTransformer  # import dopiero tu: jest wolny

        self.index = load_index(index_dir)
        self.model = SentenceTransformer(self.index.model_name)

Pamiętaj, że to, ile czasu klient da na odpowiedź, zależy od klienta, a my nie sprawdzaliśmy żadnego z popularnych edytorów ani aplikacji. Zasada jest uniwersalna: im szybciej serwer odpowie na initialize, tym mniejsze ryzyko.

Jak zmierzyć czas initialize przed publikacją serwera?

Zanim wystawisz serwer innym, zmierz, ile trwa initialize (u nas 5,4 s przy ładowaniu modelu przy imporcie, 0,3 s przy ładowaniu leniwym). Jedna linijka time w skrypcie klienta pozwala wychwycić problem, zanim zobaczą go użytkownicy.

Co klient zobaczy, gdy narzędzie zgłosi błąd?

Błędy z poprzednich części zwracaliśmy modelowi jako zwykły tekst. W MCP są dwie drogi i klient je rozróżnia. Oto, co zwrócił nasz serwer w dwóch sytuacjach:

Jak odróżnić błąd do naprawy od naruszenia schematu?

Błąd, który model może naprawić (zła nazwa sekcji), zwracaj jako tekst z listą poprawnych wartości. Naruszenie schematu zostaw FastMCP, bo oznaczy je jako błąd. W naszych testach pierwszy rodzaj wrócił jako zwykły wynik, drugi jako isError. Model zobaczył treść obu, więc w obu przypadkach może poprawić wywołanie.

Jak przetestować serwer MCP w pytest?

Nie musisz uruchamiać procesu ani klienta. Obiekt mcp ma metody list_tools, call_tool i read_resource, więc testy napiszesz bezpośrednio. Prawdziwy DocsToolbox podmieniamy atrapą z małym indeksem, żeby nie ładować modelu embeddingów:

import asyncio

import numpy as np
import pytest
from mcp.server.fastmcp.exceptions import ToolError

import server
from docs_tools import DocsToolbox
from search import Index


class FakeModel:
    def encode(self, text, normalize_embeddings=True):
        return np.array([1.0, 0.0], dtype=np.float32)


@pytest.fixture(autouse=True)
def fake_box(monkeypatch):
    box = DocsToolbox.__new__(DocsToolbox)  # bez ładowania prawdziwego modelu
    box.index = Index(
        np.array([[1.0, 0.0], [0.0, 1.0]], dtype=np.float32),
        [
            {"id": "r#0", "text": "Zwrot\n\nZwrot w 14 dni.", "source": "regulamin.md", "section": "Zwrot"},
            {"id": "r#1", "text": "Hasło\n\nReset na stronie logowania.", "source": "regulamin.md", "section": "Hasło"},
        ],
        "fake",
    )
    box.model = FakeModel()
    monkeypatch.setattr(server, "get_box", lambda: box)


def test_serwer_udostepnia_trzy_narzedzia():
    narzedzia = asyncio.run(server.mcp.list_tools())

    assert {n.name for n in narzedzia} == {"search_docs", "read_section", "list_documents"}


def test_schemat_search_docs_zawiera_ograniczenie_k():
    narzedzia = {n.name: n for n in asyncio.run(server.mcp.list_tools())}

    k = narzedzia["search_docs"].inputSchema["properties"]["k"]

    assert (k["minimum"], k["maximum"]) == (1, 5)


def test_wywolanie_search_docs_zwraca_zrodlo():
    _, wynik = asyncio.run(server.mcp.call_tool("search_docs", {"query": "zwrot"}))

    assert "regulamin.md › Zwrot" in wynik["result"]


def test_zbyt_duze_k_jest_odrzucone():
    with pytest.raises(ToolError, match="less than or equal to 5"):
        asyncio.run(server.mcp.call_tool("search_docs", {"query": "x", "k": 50}))


def test_zasob_zwraca_caly_dokument():
    tresc = asyncio.run(server.mcp.read_resource("docs://regulamin.md"))

    assert "Zwrot w 14 dni." in list(tresc)[0].content

Wszystkie pięć testów przechodzi w ułamku sekundy, bo żaden nie wymaga modelu ani sieci. Testy sprawdzają to, co dostanie klient: listę narzędzi, schemat, wynik i błąd. Jak zbudować atrapę klienta modelu językowego, wyjaśniamy w artykule o testowaniu kodu z API LLM.

Do ręcznego sprawdzania serwera istnieje też oficjalne narzędzie MCP Inspector, graficzny klient do przeglądania narzędzi i wywoływania ich z formularza. Jego instalacja i działanie zależą od wersji, więc zajrzyj do dokumentacji Model Context Protocol. My sprawdzaliśmy serwer klientem napisanym w Pythonie, którym zajmiemy się w kolejnej części.

Czy serwer MCP jest bezpieczny domyślnie?

Nie. Nasz serwer udostępnia wyłącznie narzędzia tylko do odczytu i dokumenty, które i tak są publiczne, więc ryzyko jest niewielkie. Zanim dodasz narzędzie, które coś zmienia, wróć do zasad z artykułu o uprawnieniach i zatwierdzaniu akcji agenta AI: role, tożsamość spoza modelu, zatwierdzanie przez człowieka i dziennik wywołań. W MCP zatwierdzanie jest zwykle po stronie klienta, ale to Ty odpowiadasz za to, co serwer pozwala zrobić.

Serwer uruchomiony przez streamable-http jest dostępny przez sieć, więc potrzebuje uwierzytelniania. SDK ma na to mechanizmy (parametry auth i token_verifier klasy FastMCP), ale ich konfiguracja to osobny temat, którego tu nie omawiamy. Na lokalnym komputerze serwer domyślnie nasłuchuje tylko na 127.0.0.1.

Co dalej?

Mamy działający serwer MCP z trzema narzędziami, zasobem i promptem. W następnej części napiszemy klienta MCP w Pythonie: połączymy się z tym serwerem, pobierzemy listę narzędzi, zamienimy ją na definicje dla tool callingu w OpenAI API i podłączymy do pętli agenta. Dzięki temu agent z artykułu 3 przestanie mieć narzędzia wpisane w kodzie i zacznie je dostawać od serwera.