Klient MCP w Pythonie: podłączenie serwera do agenta OpenAI

W poprzedniej części wynieśliśmy narzędzia z agenta do osobnego serwera MCP. Teraz brakuje drugiej strony: programu, który się z tym serwerem połączy, pobierze listę narzędzi i odda je agentowi. Dzięki temu agent przestaje mieć narzędzia wpisane w kodzie i dostaje je od serwera. Podmiana narzędzi nie wymaga wtedy żadnej zmiany w agencie.

To ostatnia, szósta część cyklu. Zakładam, że masz serwer MCP w Pythonie z FastMCP (SDK mcp w wersji poniżej 2, jak w poprzedniej części: pip install "mcp<2") oraz znasz pętlę agenta (agent loop) i zatwierdzanie akcji agenta.

Kto jest kim w MCP: host, klient i serwer?

W dokumentacji MCP pojawiają się trzy pojęcia i łatwo je pomylić:

Host może mieć kilka klientów naraz, po jednym na każdy serwer. My zbudujemy hosta z jednym klientem, ale kod łatwo rozszerzyć na kilka serwerów.

Jak połączyć się z serwerem MCP przez STDIO?

W trybie STDIO to klient uruchamia serwer jako proces potomny. Podajemy polecenie, a SDK zajmuje się resztą. Zapisz poniższy kod w pliku mcp_client.py:

import asyncio
import json
from contextlib import asynccontextmanager
from datetime import timedelta
from typing import Callable

from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client


@asynccontextmanager
async def connect_stdio(command: str, args: list[str], env: dict[str, str] | None = None):
    params = StdioServerParameters(command=command, args=args, env=env)
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write, read_timeout_seconds=timedelta(seconds=60)) as session:
            await session.initialize()
            yield session

Trzy elementy, które warto zrozumieć:

• sys.executable jako polecenie. Przy uruchamianiu serwera napisanego w Pythonie podaj interpreter, którego używasz (sys.executable), a nie samo python. Dzięki temu serwer dostaje to samo środowisko wirtualne z zainstalowanymi bibliotekami.
• initialize() to obowiązkowy pierwszy krok. Bez niego sesja nie działa. Tu klient i serwer wymieniają informacje o możliwościach.
• read_timeout_seconds. To limit czasu oczekiwania na odpowiedź serwera na pojedyncze żądanie. Ustawiamy 60 s, bo pierwsze wywołanie naszego serwera ładuje model embeddingów (u nas 5–7 s). Gdybyś zostawił wartość domyślną, brak limitu mógłby zawiesić agenta na zepsutym serwerze, a zbyt niski limit przerwałby pierwsze wolne wywołanie.

Dlaczego serwer nie widzi moich zmiennych środowiskowych?

To częsta pułapka. Sprawdziliśmy to na serwerze, który zwraca wartość zmiennej MOJ_KLUCZ. Ustawiliśmy ją w procesie klienta i uruchomiliśmy serwer na dwa sposoby:

Proces potomny nie dziedziczy całego środowiska klienta, więc zmienne, których serwer potrzebuje (klucze API, ścieżki), przekazujesz jawnie przez env. To zarazem zaleta: serwer nie dostaje przypadkiem sekretów, o których nic nie wie.

Jakie zmienne środowiskowe przekazać serwerowi?

Przekazuj serwerowi tylko te zmienne, których potrzebuje. Serwer, którego kodu nie napisałeś, nie dostanie przypadkiem klucza API Twojej aplikacji. Wpisuj do env wyłącznie to, co serwer faktycznie czyta.

Jak pobrać narzędzia z serwera i zamienić je na definicje dla OpenAI?

Klient pyta serwer o narzędzia (list_tools), a potem każde z nich zamieniamy na format, którego oczekuje tool calling z OpenAI. Format jest prawie taki sam jak ten z pierwszej części cyklu: nazwa, opis i schemat parametrów. Różnica polega na tym, że schemat dostajemy gotowy od serwera (inputSchema), więc nie piszemy go w Pydantic:

def to_openai_tool(tool: types.Tool) -> dict:
    return {
        "type": "function",
        "function": {
            "name": tool.name,
            "description": tool.description or "",
            "parameters": tool.inputSchema,
        },
    }


async def load_tools(session: ClientSession, allowed: set[str] | None = None):
    tools: dict[str, types.Tool] = {}
    cursor = None
    while True:
        page = await session.list_tools(cursor=cursor)
        for tool in page.tools:
            if allowed is None or tool.name in allowed:
                tools[tool.name] = tool
        cursor = page.nextCursor
        if cursor is None:
            break
    return [to_openai_tool(t) for t in tools.values()], tools

Zwracamy dwie rzeczy: listę definicji dla modelu i słownik narzędzi według nazwy, który przyda się przy wykonywaniu wywołań. Pętla z cursor obsługuje serwery, które zwracają listę narzędzi w stronach.

Parametr allowed to allowlist po stronie klienta. Serwer może po aktualizacji dodać nowe narzędzie, a Twój agent nie powinien dostać go automatycznie. Podając listę dozwolonych nazw, sam decydujesz, co agent zobaczy.

Dla naszego serwera load_tools zwróciło trzy narzędzia i zajęło razem z połączeniem 0,4 s: search_docs, read_section i list_documents. Schemat search_docs zawierał ograniczenie k od 1 do 5, które w poprzedniej części zdefiniowaliśmy przez Annotated i Field. Przeszło do modelu bez żadnej pracy po stronie klienta.

Jak wykonywać wywołania narzędzi przez sesję MCP?

Zadanie funkcji execute jest takie samo jak wcześniej: dostać nazwę i argumenty od modelu, wykonać narzędzie i zwrócić tekst. Tym razem „wykonać” znaczy wysłać żądanie do serwera, a dodatkowo pilnujemy limitu czasu i zatwierdzania:

def result_to_text(result: types.CallToolResult) -> str:
    parts = [c.text for c in result.content if c.type == "text"]
    text = "\n".join(parts) or "(narzędzie nie zwróciło tekstu)"
    return f"Błąd narzędzia: {text}" if result.isError else text


def make_execute(
    session: ClientSession,
    tools: dict[str, types.Tool],
    approve: Callable[[str, dict], bool],
    timeout: float = 30,
):
    async def execute(name: str, raw_arguments: str) -> str:
        tool = tools.get(name)
        if tool is None:
            return f"Błąd: nie ma narzędzia {name}."
        try:
            arguments = json.loads(raw_arguments or "{}")
        except json.JSONDecodeError as error:
            return f"Błąd argumentów: niepoprawny JSON ({error.msg})."

        read_only = bool(tool.annotations and tool.annotations.readOnlyHint)
        if not read_only and not approve(name, arguments):
            return "Użytkownik odrzucił tę akcję. Nie ponawiaj jej bez nowej prośby."

        try:
            result = await asyncio.wait_for(session.call_tool(name, arguments), timeout)
        except asyncio.TimeoutError:
            return f"Narzędzie {name} nie odpowiedziało w ciągu {timeout:.0f} s."
        return result_to_text(result)

    return execute

Wynik wywołania (CallToolResult) zawiera listę elementów treści, nie jeden tekst. Narzędzie MCP może zwrócić tekst, obraz lub zasób. My składamy elementy tekstowe, a jeśli nie ma żadnego, zwracamy informację zamiast pustego łańcucha. Flaga isError oznacza, że serwer zgłosił błąd, więc dodajemy prefiks „Błąd narzędzia”, żeby model wiedział, że to nie jest zwykły wynik.

Oto prawdziwe wyjścia tej funkcji, uruchomionej na serwerze z poprzedniej części:

Niepoprawny JSON i nieznane narzędzie obsługujemy w kliencie, bo do serwera w ogóle nie warto ich wysyłać. Błędy schematu (jak k=50) wykrywa serwer.

Czy zatwierdzać narzędzia MCP tak jak wcześniej?

Tak, ale klient nie zna Twoich reguł z kodu serwera. Zna za to adnotacje narzędzi, w których serwer może zadeklarować, że narzędzie tylko czyta (readOnlyHint). W poprzedniej części dopisz do narzędzi serwera adnotację:

from mcp.types import ToolAnnotations

READ_ONLY = ToolAnnotations(readOnlyHint=True)


@mcp.tool(annotations=READ_ONLY)
def search_docs(query: str, k: Annotated[int, Field(ge=1, le=5)] = 3) -> str:
    ...

Po stronie klienta zasada w execute jest prosta: narzędzie bez readOnlyHint=True traktujemy jak narzędzie zapisu i pytamy użytkownika o zgodę (tak jak Risk.WRITE w artykule o uprawnieniach). To podejście bezpieczne z założenia: gdy serwer nie zadeklaruje, że narzędzie jest tylko do odczytu, klient zakłada najgorsze.

Czy można ufać adnotacji readOnlyHint?

Adnotacja to deklaracja serwera, nie dowód. Zły albo przejęty serwer może zadeklarować readOnlyHint=True dla narzędzia, które coś kasuje. Rozstrzygaj o zatwierdzaniu po nazwie narzędzia i własnej liście allowed, a adnotacji używaj jako dodatkowego sygnału. Narzędzie bez readOnlyHint zawsze traktuj jak zapis.

Dlaczego pętla agenta musi być asynchroniczna?

SDK mcp jest asynchroniczne: list_tools i call_tool trzeba wywołać z await. Pętla agenta z drugiej części była synchroniczna, więc przepisujemy ją, a zmiany są mechaniczne:

• def run_agent zamienia się w async def run_agent,
• klient to AsyncOpenAI zamiast OpenAI,
• każde wywołanie API poprzedzamy await,
• execute zwraca wartość oczekiwaną (await execute(...)).

from openai import AsyncOpenAI


async def run_agent(
    client: AsyncOpenAI,
    question: str,
    tools: list[dict],
    execute,  # async: (nazwa, argumenty_json) -> tekst
    system_prompt: str = SYSTEM_PROMPT,
    limits: AgentLimits = AgentLimits(),
    model: str = "gpt-4o-mini",
) -> AgentResult:
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": question},
    ]
    tokens_used = 0
    seen_calls: set[tuple[str, str]] = set()

    for step in range(1, limits.max_steps + 1):
        if tokens_used >= limits.max_tokens:
            return await finish(client, messages, tools, model, step - 1, tokens_used, "max_tokens")

        response = await client.chat.completions.create(model=model, messages=messages, tools=tools)
        tokens_used += response.usage.total_tokens
        message = response.choices[0].message

        if not message.tool_calls:
            return AgentResult(message.content, step, tokens_used, "done")

        messages.append(message)
        for call in message.tool_calls:
            key = (call.function.name, call.function.arguments)
            if key in seen_calls:
                result = "To wywołanie już było. Użyj innych argumentów albo odpowiedz użytkownikowi."
            else:
                seen_calls.add(key)
                result = await execute(call.function.name, call.function.arguments)
            messages.append({"role": "tool", "tool_call_id": call.id, "content": result})

    return await finish(client, messages, tools, model, limits.max_steps, tokens_used, "max_steps")

Funkcja finish zmienia się tak samo: async def finish(...) i await client.chat.completions.create(...) z tool_choice="none". Logika limitów, wykrywania powtórzeń i kontrolowanego zakończenia jest identyczna jak w drugiej części. Ponieważ zatwierdzanie (approve) w naszym przykładzie to zwykłe input(), na czas pytania blokuje całą pętlę zdarzeń. W aplikacji konsolowej to nie przeszkadza, w serwerze webowym trzeba by je zamienić na korutynę.

Jak złożyć całość w działający program?

Wszystkie elementy łączy krótki skrypt. Klient uruchamia serwer, pobiera narzędzia, tworzy execute i oddaje je pętli agenta:

import asyncio
import sys

from openai import AsyncOpenAI

from agent_async import run_agent
from mcp_client import connect_stdio, load_tools, make_execute


def ask_user(name: str, arguments: dict) -> bool:
    print(f"Agent chce wykonać: {name} {arguments}")
    return input("Zatwierdzasz? [t/N] ").strip().lower() == "t"


async def main() -> None:
    client = AsyncOpenAI(timeout=30, max_retries=2)
    async with connect_stdio(sys.executable, ["server.py"]) as session:
        tools, by_name = await load_tools(session)
        execute = make_execute(session, by_name, ask_user)

        wynik = await run_agent(client, "Czy mogę dostać zwrot pieniędzy po tygodniu?", tools, execute)
        print(wynik.answer)
        print(wynik.steps, wynik.tokens, wynik.stop_reason)


asyncio.run(main())

Nie mogliśmy tu pokazać odpowiedzi prawdziwego modelu, bo zmienia się między uruchomieniami. Sprawdziliśmy za to cały łańcuch od strony MCP: z atrapą klienta OpenAI, która zwracała ustalone wywołania, agent połączył się z prawdziwym serwerem przez STDIO, pobrał trzy narzędzia, wykonał search_docs, a potem dwa wywołania read_section naraz (jedno z nieistniejącą sekcją, co serwer obsłużył komunikatem z listą dostępnych) i zakończył w trzech krokach ze stop_reason == "done". Żadnego z trzech narzędzi nie musiał zatwierdzać, bo wszystkie mają readOnlyHint.

Jak połączyć się z serwerem MCP przez HTTP?

Dla serwera uruchomionego jako usługa HTTP (python server.py streamable-http) zmieniasz tylko sposób nawiązania połączenia. Reszta kodu (ClientSession, load_tools, make_execute) pozostaje taka sama:

from mcp.client.streamable_http import streamablehttp_client


@asynccontextmanager
async def connect_http(url: str, headers: dict[str, str] | None = None):
    async with streamablehttp_client(url, headers=headers) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            yield session

Częsty błąd dotyczy adresu. Adres serwera musi wskazywać punkt końcowy MCP, domyślnie /mcp. Sprawdziliśmy trzy adresy na naszym serwerze:

Komunikat „Session terminated” nic nie mówi o przyczynie, więc przy takim błędzie najpierw sprawdź adres i transport serwera. Podobne pytania o błąd 404 przy pobieraniu narzędzi z serwera MCP pojawiają się na Stack Overflow, a jedną z możliwych przyczyn jest właśnie niewłaściwa ścieżka.

Parametr headers pozwala dodać nagłówek Authorization: Bearer … do każdego żądania. To tylko połowa sprawy, bo serwer musi ten token zweryfikować. Konfiguracja uwierzytelniania po stronie serwera wykracza poza ten cykl.

Jak przetestować klienta MCP w pytest?

Nie potrzebujesz procesu potomnego. SDK udostępnia połączenie klienta z serwerem w pamięci (create_connected_server_and_client_session), więc testy są szybkie i nie zostawiają procesów. Do testów tworzymy mały serwer demo z trzema narzędziami: odczytem z readOnlyHint, kasowaniem bez adnotacji (wymaga zatwierdzenia) i narzędziem celowo wolnym:

import asyncio
import json

from mcp.server.fastmcp import FastMCP
from mcp.shared.memory import create_connected_server_and_client_session
from mcp.types import ToolAnnotations

from mcp_client import load_tools, make_execute

demo = FastMCP("demo")
wywolane: list[int] = []


@demo.tool(annotations=ToolAnnotations(readOnlyHint=True))
def czytaj(numer: int) -> str:
    """Czyta notatkę o podanym numerze."""
    return f"notatka {numer}"


@demo.tool()
def usun(numer: int) -> str:
    """Usuwa notatkę o podanym numerze."""
    wywolane.append(numer)
    return f"usunięto {numer}"


@demo.tool(annotations=ToolAnnotations(readOnlyHint=True))
async def wolne() -> str:
    """Narzędzie, które odpowiada za wolno."""
    await asyncio.sleep(1)
    return "gotowe"


def z_serwerem(test):
    """Uruchamia test z klientem połączonym z serwerem w pamięci, bez procesu potomnego."""
    async def uruchom():
        wywolane.clear()
        async with create_connected_server_and_client_session(demo._mcp_server) as session:
            await test(session)

    asyncio.run(uruchom())


def test_narzedzia_serwera_zamieniaja_sie_na_definicje_openai():
    async def test(session):
        tools, _ = await load_tools(session)

        czytaj = next(t for t in tools if t["function"]["name"] == "czytaj")
        assert czytaj["type"] == "function"
        assert czytaj["function"]["description"] == "Czyta notatkę o podanym numerze."
        assert czytaj["function"]["parameters"]["required"] == ["numer"]

    z_serwerem(test)


def test_narzedzie_bez_readonlyhint_wymaga_zatwierdzenia():
    async def test(session):
        _, by_name = await load_tools(session)
        execute = make_execute(session, by_name, lambda n, a: False)

        wynik = await execute("usun", json.dumps({"numer": 3}))

        assert wynik.startswith("Użytkownik odrzucił")
        assert wywolane == []  # serwer w ogóle nie dostał wywołania

    z_serwerem(test)


def test_wolne_narzedzie_przekracza_limit_czasu():
    async def test(session):
        _, by_name = await load_tools(session)
        execute = make_execute(session, by_name, lambda n, a: True, timeout=0.1)

        assert "nie odpowiedziało" in await execute("wolne", "{}")

    z_serwerem(test)

W pełnym zestawie mamy osiem testów (dodatkowo: filtr allowed, odczyt bez pytania, zatwierdzone narzędzie, niepoprawny JSON z nieznanym narzędziem oraz błąd walidacji serwera) i wszystkie przechodzą w niecałą sekundę. Najważniejszy jest test odrzucenia: sprawdza nie tylko komunikat dla modelu, ale też to, że serwer w ogóle nie dostał wywołania. Jak podstawić atrapę klienta modelu językowego i przetestować samą pętlę agenta, pokazujemy w artykule o testowaniu kodu z API LLM.

Co dalej po całym cyklu?

Przeszliśmy całą drogę: od pierwszego wywołania narzędzia przez model, przez pętlę agenta z limitami, agenta czytającego dokumenty i zasady uprawnień, aż po serwer i klienta MCP. Ten sam agent może teraz dostawać narzędzia od dowolnego serwera MCP, a Ty kontrolujesz, które z nich zobaczy i które wymagają zatwierdzenia.

Naturalne kolejne kroki to podłączenie kilku serwerów naraz (po jednym kliencie na serwer), uwierzytelnianie zdalnego serwera i zamiana prostego input() na prawdziwy interfejs zatwierdzania. Jeśli chcesz zbudować taki projekt z mentorem i dostać code review swojego kodu, zajrzyj na ścieżkę Python.