Uprawnienia agenta AI w Pythonie i zatwierdzanie jego akcji

Agent z poprzedniej części tylko czytał dokumenty. Najgorsze, co mógł zrobić, to udzielić złej odpowiedzi. Zupełnie inaczej wygląda sytuacja, gdy dostaje narzędzie, które coś zmienia: składa wniosek o zwrot pieniędzy, wysyła maila, usuwa rekord. Wtedy pomyłka modelu albo cudze polecenie ukryte w danych kończy się realną szkodą.

To czwarta część cyklu. Zakładam, że znasz agenta z narzędziami wyszukiwania i odczytu dokumentów, jego pętlę (agent loop) i tool calling w OpenAI API. Rozbudujemy go o dwa narzędzia sklepu (podgląd zamówienia i wniosek o zwrot), role użytkowników, zatwierdzanie przez człowieka i dziennik wywołań. Sama pętla agenta nie zmieni się ani o linijkę.

Dlaczego narzędzia zapisu wymagają osobnych zasad?

Narzędzie tylko do odczytu można wywołać pomyłkowo i nic się nie stanie. Narzędzie zapisu już nie. Dlatego dzielimy narzędzia według skutków:

W naszym kodzie wystarczą dwa poziomy: READ i WRITE. Jeśli masz narzędzia nieodwracalne, najbezpieczniej w ogóle nie dawać ich agentowi.

Skąd mogą się wziąć złe wywołania narzędzi?

Są trzy źródła i każde wymaga innej obrony:

Trzecie źródło nazywa się prompt injection (wstrzyknięcie polecenia). Model nie potrafi niezawodnie odróżnić instrukcji od tekstu, który ma tylko przetworzyć, więc każdy tekst trafiający do kontekstu (wynik wyszukiwania, treść strony, mail) może próbować nim sterować. Nie da się tego w pełni wykluczyć promptem, dlatego zabezpieczenia budujemy w kodzie wokół modelu: nawet jeśli model da się namówić na złe wywołanie, kod odmówi jego wykonania.

Jak opisać narzędzia razem z poziomem ryzyka i rolami?

Rejestr narzędzi z poprzednich części trzymał tylko model argumentów i funkcję. Rozszerzamy go o poziom ryzyka i listę ról, które mogą narzędzie wywołać. Zapisz ten i kolejne fragmenty w pliku policy.py, obok docs_tools.py i agent.py:

from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum
from typing import Callable

from pydantic import BaseModel, Field, ValidationError

from docs_tools import DocsToolbox, ListDocuments, ReadSection, SearchDocs, to_tool_definition


class Risk(str, Enum):
    READ = "read"
    WRITE = "write"


@dataclass(frozen=True)
class User:
    id: str
    role: str  # "gosc" albo "klient"


@dataclass
class ToolSpec:
    model: type[BaseModel]
    function: Callable[[BaseModel, User], str]
    risk: Risk
    roles: frozenset[str]

Funkcje narzędzi dostają teraz drugi argument: User. To on, a nie model, mówi, w czyim imieniu agent działa.

Dlaczego tożsamość bierzemy z sesji, a nie od modelu?

Zobacz narzędzia sklepu. Argumentem jest tylko numer zamówienia, bez żadnego identyfikatora klienta:

class GetOrder(BaseModel):
    """Pokazuje zamówienie zalogowanego klienta: kwotę, datę płatności i status."""

    order_id: str = Field(pattern=r"^ZAM-\d{4}$", description="Numer zamówienia, np. 'ZAM-1001'")


class RequestRefund(BaseModel):
    """Składa wniosek o zwrot pieniędzy za zamówienie zalogowanego klienta.
    Zmienia dane i wymaga zatwierdzenia przez klienta."""

    order_id: str = Field(pattern=r"^ZAM-\d{4}$", description="Numer zamówienia, np. 'ZAM-1001'")
    reason: str = Field(max_length=200, description="Powód zwrotu podany przez klienta")


class Shop:
    def __init__(self):
        self.orders = {
            "ZAM-1001": {"user": "u1", "amount": 49.0, "days_ago": 5, "started": False, "status": "opłacone"},
            "ZAM-1002": {"user": "u1", "amount": 490.0, "days_ago": 40, "started": True, "status": "opłacone"},
            "ZAM-2001": {"user": "u2", "amount": 49.0, "days_ago": 3, "started": False, "status": "opłacone"},
        }

    def _own_order(self, order_id: str, user: User) -> dict | None:
        order = self.orders.get(order_id)
        return order if order and order["user"] == user.id else None  # id z sesji, nie od modelu

    def get_order(self, args: GetOrder, user: User) -> str:
        order = self._own_order(args.order_id, user)
        if order is None:
            return f"Nie znaleziono zamówienia {args.order_id}."
        started = "tak" if order["started"] else "nie"
        return (
            f"{args.order_id}: {order['amount']} zł, opłacone {order['days_ago']} dni temu, "
            f"rozpoczęty kurs: {started}, status: {order['status']}."
        )

    def request_refund(self, args: RequestRefund, user: User) -> str:
        order = self._own_order(args.order_id, user)
        if order is None:
            return f"Nie znaleziono zamówienia {args.order_id}."
        if order["days_ago"] > 14 or order["started"]:
            return "Zwrot niemożliwy: minęło 14 dni od płatności albo kurs został rozpoczęty."
        order["status"] = "zwrot zgłoszony"
        return f"Zgłoszono zwrot {order['amount']} zł. Środki wrócą w ciągu 7 dni roboczych."

Trzy rzeczy zasługują na uwagę:

• Brak pola user_id w argumentach. Gdyby go tam było, model (albo ukryte w danych polecenie) mógłby wpisać cudzy identyfikator. Teraz nie ma co podmieniać.
• Ten sam komunikat dla zamówienia nieistniejącego i cudzego. „Nie znaleziono zamówienia” nie zdradza, czy numer należy do kogoś innego.
• Reguły biznesowe sprawdzamy w kodzie. Kwotę zwrotu i warunki (14 dni, brak rozpoczętego kursu, zgodnie z regulaminem z poprzednich części) liczy request_refund, a nie model. Model nie podaje kwoty, więc nie może jej zawyżyć.

Jak zbudować rejestr i narzędzia dla konkretnego użytkownika?

Rejestr łączy wszystkie narzędzia: te z poprzedniej części (odczyt dokumentów) i nowe, z ryzykiem i rolami:

ALL_ROLES = frozenset({"gosc", "klient"})
CLIENT_ONLY = frozenset({"klient"})


def build_registry(box: DocsToolbox, shop: Shop) -> dict[str, ToolSpec]:
    return {
        "search_docs": ToolSpec(SearchDocs, lambda a, u: box.search_docs(a), Risk.READ, ALL_ROLES),
        "read_section": ToolSpec(ReadSection, lambda a, u: box.read_section(a), Risk.READ, ALL_ROLES),
        "list_documents": ToolSpec(ListDocuments, lambda a, u: box.list_documents(a), Risk.READ, ALL_ROLES),
        "get_order": ToolSpec(GetOrder, shop.get_order, Risk.READ, CLIENT_ONLY),
        "request_refund": ToolSpec(RequestRefund, shop.request_refund, Risk.WRITE, CLIENT_ONLY),
    }

Teraz najważniejsza funkcja. build_agent_tools zwraca, tak jak w poprzedniej części, listę definicji i funkcję execute, ale tym razem dla konkretnego użytkownika. Dodatkowo przyjmuje funkcję approve, która pyta człowieka o zgodę, oraz listę audit na dziennik wywołań:

def build_agent_tools(registry, user: User, approve: Callable[[str, BaseModel], bool], audit: list[dict]):
    allowed = {name: spec for name, spec in registry.items() if user.role in spec.roles}
    tools = [to_tool_definition(name, spec.model) for name, spec in allowed.items()]

    def record(name: str, raw: str, decision: str, result: str) -> str:
        audit.append({
            "time": datetime.now(timezone.utc).isoformat(timespec="seconds"),
            "user": user.id, "tool": name, "arguments": raw, "decision": decision, "result": result,
        })
        return result

    def execute(name: str, raw_arguments: str) -> str:
        if name not in allowed:  # także gdy model "wymyślił" narzędzie, którego nie dostał
            return record(name, raw_arguments, "brak uprawnień", "Brak uprawnień do tego narzędzia.")
        spec = allowed[name]
        try:
            args = spec.model.model_validate_json(raw_arguments or "{}")
        except ValidationError as error:
            return record(name, raw_arguments, "błąd argumentów", f"Błąd argumentów: {error}")

        if spec.risk is Risk.WRITE and not approve(name, args):
            return record(
                name, raw_arguments, "odrzucone",
                "Użytkownik odrzucił tę akcję. Nie ponawiaj jej bez nowej prośby.",
            )
        decision = "zatwierdzone" if spec.risk is Risk.WRITE else "automatycznie"
        return record(name, raw_arguments, decision, spec.function(args, user))

    return tools, execute

W tej jednej funkcji działają dwie niezależne warstwy ograniczeń uprawnień:

1. Filtr listy narzędzi. Gość w ogóle nie dostaje definicji get_order ani request_refund, więc model nie wie, że istnieją. To w praktyce allowlist: lista narzędzi dozwolonych dla danej roli, a wszystko spoza niej jest zabronione.
2. Sprawdzenie w execute. Gdyby model mimo to poprosił o niedozwolone narzędzie (bo zgadł nazwę albo ktoś go do tego namówił), execute odmówi.

Pierwsza warstwa zmniejsza pokusę, druga jest właściwym zabezpieczeniem. Nigdy nie polegaj tylko na tym, że „model nie widzi narzędzia”.

Jak zatwierdzać akcje przez człowieka?

Zatwierdzanie przez człowieka (po angielsku human in the loop) opiera się na funkcji approve, która dostaje nazwę narzędzia i zwalidowane argumenty i zwraca True albo False. W aplikacji konsolowej wystarczy input():

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

W aplikacji webowej ta sama funkcja zapisze prośbę o zatwierdzenie i wstrzyma agenta do kliknięcia przycisku, ale zasada jest ta sama: pętla agenta nie wie nic o zatwierdzaniu. Dla niej to po prostu narzędzie, które odpowiada wolniej.

Uruchomienie wygląda tak:

box = DocsToolbox()
shop = Shop()
registry = build_registry(box, shop)
audit: list[dict] = []

user = User("u1", "klient")  # w prawdziwej aplikacji z sesji logowania
tools, execute = build_agent_tools(registry, user, ask_user, audit)

wynik = run_agent(client, "Chcę zwrot za zamówienie ZAM-1001", tools, execute)
print(wynik.answer)

Zwróć uwagę, że run_agent dostaje tools i execute dokładnie tak, jak w poprzedniej części. Uprawnienia i zatwierdzanie są całkowicie poza pętlą.

Co użytkownik powinien zobaczyć przy zatwierdzaniu?

Pokazuj argumenty, które faktycznie zostaną wykonane, a nie opis napisany przez model. Opis może być mylący (lub zmanipulowany), argumenty to dokładnie to, co wejdzie do funkcji. W naszym ask_user użytkownik widzi nazwę narzędzia i słownik z order_id oraz reason, więc wie, co zatwierdza.

Co zwrócić modelowi, gdy użytkownik odmówi?

Odmowa to zwykły wynik narzędzia. API wymaga, żeby na każde tool_call_id odpowiedzieć wiadomością tool, więc nie możesz po prostu pominąć wywołania. Zwracamy tekst: „Użytkownik odrzucił tę akcję. Nie ponawiaj jej bez nowej prośby.” Dzięki temu model:

• wie, że akcja się nie wykonała, i nie napisze użytkownikowi, że zwrot został złożony,
• nie próbuje jej ponawiać w kolejnych krokach (a gdyby próbował, wykrywanie powtórzeń w pętli z drugiej części zadziała jako druga linia obrony).

Wynik zatwierdzenia i faktycznego wykonania to dwie różne rzeczy. Zatwierdzenie znaczy tylko „wolno próbować”. Reguły biznesowe nadal sprawdza funkcja narzędzia.

Czy zgoda użytkownika wystarcza, żeby wykonać akcję?

Nie. Zgoda użytkownika oznacza „wolno spróbować”, a nie „na pewno wykonaj”. W naszym teście zatwierdzony zwrot za zamówienie sprzed 40 dni został odrzucony przez regułę 14 dni, a zatwierdzony zwrot cudzego zamówienia przez sprawdzenie właściciela. Nie przenoś logiki biznesowej do okna zatwierdzania.

Co dostaje agent w typowych sytuacjach?

Oto prawdziwe wyjścia naszego kodu (indeks dokumentów z modułu 6, trzy zamówienia, dwóch klientów):

Wiersze z cudzym zamówieniem i zwrotem po 40 dniach pokazują, dlaczego sama zgoda użytkownika nie wystarcza: nawet zatwierdzone żądanie przechodzi przez reguły biznesowe i sprawdzenie właściciela. W drugim wierszu zwrot rzeczywiście się wykonał i zmienił status zamówienia, w trzecim zamówienie zostało nietknięte.

Jak chronić się przed poleceniami ukrytymi w danych?

Wyniki narzędzi to dane, a nie polecenia. Tak samo jak przy XSS czy SQL injection problem polega na tym, że tekst z zewnątrz trafia do miejsca, w którym może zostać potraktowany jak instrukcja. Różnica: tutaj nie ma odpowiednika zapytań parametryzowanych, który rozdzieli dane od poleceń w sposób niezawodny. Zostaje obrona warstwowa:

Żadna z tych warstw nie zakłada, że model jest odporny na manipulację. Zakładamy, że czasem da się go nakłonić do złego wywołania, i dlatego każde takie wywołanie musi przejść przez kod, który nie ufa modelowi. Narzędzia czytające zewnętrzne treści (strony, maile) łączone z narzędziami zapisu to szczególnie ryzykowne połączenie, więc w takim agencie zatwierdzenie jest obowiązkowe.

Jak sprawdzić, czy agent nie zmyślił źródła odpowiedzi?

Agent z poprzedniej części ma w prompcie podawać źródło w formacie (plik › sekcja). Zanim pokażesz odpowiedź użytkownikowi, sprawdź w kodzie, czy to źródło wystąpiło w wynikach któregoś narzędzia podczas tej rozmowy. Jeśli model podał źródło, którego nie widział, to sygnał, że je wymyślił, i odpowiedź warto odrzucić.

Jak przetestować uprawnienia w pytest?

Uprawnienia są kodem deterministycznym, więc testuje się je łatwo i bez modelu. Rejestr tworzymy z atrapą zamiast prawdziwych embeddingów (wyszukiwarka nie jest tu testowana):

import json

import numpy as np
import pytest

from agent import run_agent
from docs_tools import DocsToolbox
from policy import Shop, User, build_agent_tools, build_registry
from search import Index


@pytest.fixture
def registry():
    box = DocsToolbox.__new__(DocsToolbox)  # bez ładowania modelu embeddingów
    box.index = Index(np.array([[1.0, 0.0]], dtype=np.float32), [], "fake")
    return build_registry(box, Shop())


def zbuduj(registry, user, approve=lambda nazwa, args: True):
    audit: list[dict] = []
    tools, execute = build_agent_tools(registry, user, approve, audit)
    return tools, execute, audit


def test_gosc_nie_dostaje_narzedzi_zapisu(registry):
    tools, _, _ = zbuduj(registry, User("g", "gosc"))

    assert "request_refund" not in [t["function"]["name"] for t in tools]


def test_gosc_nie_wykona_narzedzia_nawet_gdy_model_o_nie_poprosi(registry):
    _, execute, audit = zbuduj(registry, User("g", "gosc"))

    wynik = execute("request_refund", '{"order_id": "ZAM-1001", "reason": "x"}')

    assert wynik == "Brak uprawnień do tego narzędzia."
    assert audit[-1]["decision"] == "brak uprawnień"


def test_odrzucona_akcja_nie_zmienia_danych(registry):
    _, execute, audit = zbuduj(registry, User("u1", "klient"), approve=lambda n, a: False)

    wynik = execute("request_refund", '{"order_id": "ZAM-1001", "reason": "x"}')

    assert wynik.startswith("Użytkownik odrzucił")
    assert audit[-1]["decision"] == "odrzucone"
    assert "status: opłacone" in execute("get_order", '{"order_id": "ZAM-1001"}')


def test_klient_nie_zwroci_cudzego_zamowienia(registry):
    _, execute, _ = zbuduj(registry, User("u1", "klient"))

    wynik = execute("request_refund", '{"order_id": "ZAM-2001", "reason": "x"}')

    assert wynik == "Nie znaleziono zamówienia ZAM-2001."

Dwa ostatnie testy pokazują, że uprawnienia nie zależą od zachowania modelu: sprawdzamy bezpośrednio execute, czyli to, co zrobi kod, gdy model poprosi o cokolwiek. Dodaj jeszcze test z atrapą klienta (jak w poprzedniej części), w którym „model” prosi o zwrot, a funkcja approve odmawia. Taki test potwierdza, że pętla kończy się normalnie (stop_reason == "done"), a model dostaje informację o odmowie. Technikę atrap klienta opisujemy w artykule o testowaniu kodu z API LLM.

Co dalej?

Mamy agenta, który zna swoje granice. Dotychczas jednak każde narzędzie żyło w tym samym procesie Pythona co agent. W następnej części wyniesiemy narzędzia poza program: napiszemy serwer MCP w Pythonie (Model Context Protocol), do którego może podłączyć się dowolny klient, a w kolejnej klienta MCP w Pythonie, który użyje takich narzędzi w agencie. Jeśli chcesz zobaczyć, jak w aplikacji webowej chroni się endpointy przed niepowołanym dostępem, zajrzyj do artykułu o uwierzytelnianiu w FastAPI.