Dodaj DocketProof do agenta LangGraph

Co właściwie dodajesz

@guarded_tool owija jedną funkcję Pythona — zwykle węzeł narzędzia w LangGraph — lokalną, zadeklarowaną kontrolą polityki oraz podpisanym, dopisywalnym łańcuchem wpisów. Werdykt allow uruchamia funkcję od razu i zapisuje podpisany wpis typu action. Werdykt deny nigdy nie uruchamia funkcji i zamiast tego zapisuje podpisany wpis typu denial. Werdykt review_required zatrzymuje bieżący przebieg grafu LangGraph przez wbudowany interrupt(), aż człowiek go wznowi — patrz niżej.

Instalacja

docketproof nie został jeszcze opublikowany na PyPI, więc na razie instaluje się go z lokalnej kopii pakietu (katalog docketproof/ w jego repozytorium):

$ pip install -e .

Zdefiniuj politykę

Polityka to dokładne mapowanie nazwa-narzędzia → werdykt. Nie ma tu symboli wieloznacznych ani języka wyrażeń — każde chronione narzędzie potrzebuje własnego, nazwanego wpisu, a narzędzie bez wpisu jest oceniane jako deny, nie allow: brak wpisu w polityce to nie zgoda. Jako plik YAML:

delete_resource: review_required
read_resource: allow
wipe_everything: deny

Albo zbudowana bezpośrednio jako słownik Pythona, przez MandatePolicy.from_dict — pokazane w przykładzie w następnej sekcji.

Owiń narzędzie

Przekaż dokładnie jeden z parametrów: policy (raz zbudowany obiekt MandatePolicy) albo policy_path (ścieżkę do pliku, wczytywaną i walidowaną ponownie przy każdym wywołaniu — przydatne, gdy operator musi zmienić politykę bez restartu procesu). chain_path jest zawsze wymagany: to tam trafia podpisany łańcuch wpisów.

from docketproof import MandatePolicy, guarded_tool

policy = MandatePolicy.from_dict({
    "delete_resource": "review_required",
    "read_resource": "allow",
    "wipe_everything": "deny",
})

@guarded_tool(policy=policy, chain_path="./mandate-chain.jsonl")
def delete_resource(resource_id: str) -> str:
    ...  # tu Twoja właściwa logika narzędzia, bez zmian
    return f"deleted {resource_id}"

Aby zamiast tego wczytywać plik polityki przy każdym wywołaniu i być powiadamianym, gdy sama bramka się psuje:

@guarded_tool(
    policy_path="./policy.yaml",   # wczytywany i walidowany ponownie przy każdym wywołaniu
    chain_path="./mandate-chain.jsonl",
    on_failure_webhook="https://alerts.example.com/gate-failures",
)
def delete_resource(resource_id: str) -> str:
    ...

Narzędzia asynchroniczne

guarded_tool automatycznie wykrywa narzędzie async def — przez inspect.iscoroutinefunction — i zwraca pasujący, asynchroniczny wrapper. Bez dodatkowego parametru, bez osobnej nazwy dekoratora. Wszystkie gwarancje z powyżej są identyczne w obu przypadkach: polityka jest oceniana przed uruchomieniem narzędzia, bramka blokuje się przy tych samych błędach mechanizmu, a hostowana synchronizacja zachowuje się tak samo:

@guarded_tool(policy=policy, chain_path="./mandate-chain.jsonl")
async def delete_resource(resource_id: str) -> str:
    ...  # tu Twoja właściwa logika narzędzia (async), bez zmian
    return f"deleted {resource_id}"

# Wywołane z asynchronicznego węzła LangGraph:
result = await delete_resource(resource_id="res_42")

Wywołuj chronioną funkcję bezpośrednio przez await, albo przez coś, co samo sprawdza, czy dany obiekt jest funkcją korutynową, zanim zdecyduje, czy wywołać go z await — tak właśnie LangGraph, i większość asynchronicznych narzędzi w Pythonie, odróżnia wywołania synchroniczne od asynchronicznych. Wywołujący, który zakłada synchroniczność bez sprawdzenia, nie uruchomi tego poprawnie niezależnie od tego, czym to narzędzie jest owinięte.

Co dzieje się przy review_required

Graf zatrzymuje się przez interrupt() z LangGraph, udostępniając słownik z polami tool_name, args i policy_name temu, co steruje grafem. Wznów przebieg za pomocą Command(resume=...), korzystając z dwóch pomocniczych funkcji z tego pakietu. Działa to tak samo niezależnie od tego, czy chronione narzędzie jest synchroniczne czy asynchroniczne — interrupt() to zwykłe, synchroniczne wywołanie wewnątrz obu wrapperów i nie potrzebuje własnego await:

from docketproof import Command, resume_approved, resume_denied

# W kodzie, który steruje grafem (CLI, handler serwera, konsola
# human-in-the-loop), gdy już zdecydujesz, co zrobić z zatrzymanym
# wywołaniem:
graph.invoke(Command(resume=resume_approved("looks right")), config)
# albo:
graph.invoke(Command(resume=resume_denied("not today")), config)

Niezależnie od decyzji człowieka, powstały wpis — action dla zatwierdzenia, denial dla odrzucenia — otrzymuje ten sam podpis i to samo traktowanie w łańcuchu. Nie ma osobnego, mniej ważnego zapisu dla odrzucenia: to właśnie on dowodzi, że blokada faktycznie zadziałała.

Gdzie trafia lokalny plik rejestru

chain_path to zwykły plik JSON Lines — jeden podpisany wpis na linię, tylko do dopisywania. Jeśli nie przekażesz własnego keypair=, klucz podpisujący jest generowany raz i przechowywany obok pliku jako <chain_path>.key (uprawnienia pliku 0600). Uruchom docketproof doctor przed pierwszą prawdziwą sesją — i ponownie w CI albo skryptach startowych — aby wykryć uszkodzony klucz, niezapisywalny rejestr albo błędną politykę, zanim zacznie to po cichu blokować każde wywołanie:

$ docketproof doctor --docket-path ./mandate-chain.jsonl --policy ./policy.yaml

Nie masz jeszcze klucza podpisującego? Wygeneruj go bezpośrednio:

$ docketproof init-key --docket-path ./mandate-chain.jsonl

Opcjonalnie: podłącz hostowany dashboard

Przekaż razem hosted_api_key i hosted_api_url, aby każdy podpisany wpis był kopiowany do hostowanego dashboardu — dopiero po tym, jak trafił już do lokalnego łańcucha. To mechanizm, który działa w tle i nie czeka na potwierdzenie: lokalny zapis jest zawsze rozstrzygający, a wynik owiniętego wywołania, zgłoszony wyjątek i zawartość lokalnego łańcucha są identyczne niezależnie od tego, czy synchronizacja się uda, zawiedzie, czy w ogóle nie jest skonfigurowana:

import os

@guarded_tool(
    policy=policy,
    chain_path="./mandate-chain.jsonl",
    hosted_api_key=os.environ.get("DOCKETPROOF_API_KEY"),
    hosted_api_url=os.environ.get("DOCKETPROOF_API_URL"),
)
def delete_resource(resource_id: str) -> str:
    ...

Nie ma tu odpytywania o zdalną decyzję człowieka — review_required jest zawsze rozstrzygany lokalnie, przez interrupt()/Command(resume=...) powyżej. Hostowana synchronizacja tylko kopiuje już zakończony wpis, później.

Zweryfikuj łańcuch niezależnie

Wyeksportuj łańcuch, do którego pisze twoje chronione narzędzie, i sprawdź go osobnym, niezależnie zaimplementowanym weryfikatorem w TypeScripcie — innym kodem niż ten, który go podpisał:

chain = delete_resource.mandate_chain   # LocalChain, do którego pisze to narzędzie
bundle = chain.to_bundle()

import json
open("bundle.json", "w").write(json.dumps(bundle, indent=2))

$ mandate-verify bundle.json

Co dowodzi wpis — i czego nie dowodzi

To to samo zastrzeżenie zakresu, co w sekcji na stronie głównej — pełny kontekst na docketproof.dev/pl/#scope.

Czego ten poradnik nie obejmuje

Ten pakiet nie ma języka wyrażeń dla polityk (tylko dokładne dopasowanie nazwy narzędzia), nie ma budżetów ani limitów wydatków, nie ocenia ryzyka na podstawie argumentów wywołania i nie wspiera innych frameworków agentowych niż LangGraph. Żadne z tych ograniczeń nie jest planowane jako rozszerzenie akurat tego dekoratora — pełną listę tego, czego pakiet celowo nie robi, znajdziesz w jego README.