Chroń komunikację agentów przez ACP¶
FastFence przyjmuje żądania Agent Communication Protocol i przekazuje je do skonfigurowanego agenta przez istniejący silnik polityk narzędzi:
ACP client → FastFence /acp/runs → input controls → trusted ACP peer
↓
ACP response ← output controls ← completed peer response
Jest to protokół REST IBM/BeeAI, odrębny od Agent Client Protocol dla edytorów. Jego oficjalne repozytorium zostało zarchiwizowane, a projekt przeszedł do A2A. FastFence implementuje ograniczony profil zgodności: synchroniczne, bezstanowe wywołania z jawnym tekstem w wiadomości i uwierzytelnione wykrywanie agentów. Nie deklaruje pełnej zgodności z ACP ani A2A.
Uruchom rzeczywistego agenta lokalnie¶
Zainstaluj FastFence 1.0.0 lub nowszy według Pierwszych kroków, a następnie rozpakuj kompletne archiwum przykładów do examples/. Wszystkie polecenia wykonuj z katalogu instalacji. Bramkę uruchom we własnym środowisku FastFence przez uv run. Utwórz osobne środowisko dla agenta i klienta zarchiwizowanego SDK; przypięta wersja Uvicorn nie zmienia wtedy bramki. SDK importuje też requests, nie deklarując tej zależności, dlatego zainstaluj ją jawnie w tym osobnym środowisku:
uv venv --python 3.12 .acp-venv
uv pip install --python .acp-venv/bin/python 'acp-sdk==1.0.3' 'uvicorn==0.35.0' 'requests==2.34.2'
.acp-venv/bin/python examples/acp_server.py
Oficjalny ACP SDK udostępnia rzeczywistą operację zamiany na wielkie litery na lokalnym porcie 8020. Generuje osobny prywatny token backendu w state/examples/acp-upstream-token.txt. Każdy endpoint backendu wymaga tego tokenu. Zawartość pliku nigdy nie jest wypisywana.
W drugim terminalu uruchom bramkę z jej własnymi zależnościami FastFence:
To uruchamia odizolowaną instancję FastFence na porcie 8030 i rejestruje agenta uppercase jako narzędzie polityki acp.uppercase. Konfiguracja i tokeny bramki znajdują się pod state/examples/acp-gateway/. Ponowny start zachowuje istniejącą konfigurację i nie edytuje głównej instalacji. Jawna polityka przykładu używa kontroli deterministycznych, więc nie wymaga modelu; domyślna polityka produktu nadal wymaga Laya.
W trzecim terminalu użyj oficjalnego klienta ACP SDK:
.acp-venv/bin/python examples/acp_client.py --prompt 'hello'
.acp-venv/bin/python examples/acp_client.py --prompt 'forbidden'
.acp-venv/bin/python examples/acp_client.py --prompt '[email protected]'
Oczekiwany wynik: hello kończy się wynikiem HELLO; forbidden zwraca nieudane wykonanie przed uruchomieniem agenta, a skrypt kończy się kodem 1; adres e-mail jest redagowany przed przekazaniem. HTTP 200 może zawierać nieudane wykonanie: sprawdź status, error.data.reason i error.data.upstream_executed. Poprawny wynik jest zwracany dopiero po przejściu kontroli wyjścia. Sprawdź decyzję w Activity pod http://127.0.0.1:8030. UUID wykonania ACP odpowiada identyfikatorowi żądania w audycie bez myślników UUID.
Serwer i bramka przyjmują --port; bramka dodatkowo --upstream-url. Klient przyjmuje --url, --agent i --credentials. Domyślnym tokenem jest local-agent odizolowanego przykładu; tokeny administracyjne nie mogą wywoływać ACP.
Podłącz istniejącego agenta ACP¶
Skonfiguruj zaufane ustawienia startowe w środowisku bramki lub prywatnym .env:
export FASTFENCE_ACP_AGENTS='{"assistant":{"base_url":"https://peer.example.org","agent_name":"assistant"}}'
Zastąp przykładowy URL i nazwę agenta własnymi wartościami. Dodaj api_key do tej zaufanej konfiguracji, jeśli agent wymaga tokenu Bearer. Jest to token backendu, osobny od tokenów FastFence używanych przez klientów. Nie jest zwracany przez API wykrywania agentów ani polityk. Zdalni agenci wymagają HTTPS; HTTP jest dopuszczony tylko dla adresów loopback. Dane klienta nie mogą wybierać adresów upstream ani ich tokenów.
Dodaj odpowiednie narzędzie do config/policy.yaml, zachowując pozostałe ustawienia i zwiększając aktywną wersję polityki:
Po zmianie ustawień startowych uruchom proces ponownie. Zmiany polityki nadal przeładowują się na żywo. Przy zwykłej bramce działającej na porcie 8000:
.acp-venv/bin/python examples/acp_client.py --url http://127.0.0.1:8000 \
--credentials state/credentials.json --agent assistant --prompt 'Hello'
GET /acp/agents pokazuje wyłącznie agentów zarejestrowanych, skonfigurowanych i dozwolonych dla roli. POST /acp/runs przyjmuje tę samą tożsamość Bearer co REST/MCP; wszystkie te transporty współdzielą politykę i budżet w pamięci dla danego podmiotu. Koszty agenta to skonfigurowany koszt narzędzia i ograniczone rozliczanie zasobów tekstowych, a nie pomiary tokenów rozliczeniowych dostawcy. Bezpośredni dostęp do backendu musi pozostawać ograniczony do zaufanych tokenów bramki lub właściwych granic sieciowych.
Kontrole obejmują wiadomości przechodzące przez FastFence. Nie sprawdzają wewnętrznych wywołań modeli/narzędzi zdalnego agenta, chyba że one również przechodzą przez FastFence; ukryte wewnętrzne zużycie tokenów nie jest mierzone przez ten adapter.
Obsługiwana treść i wykonanie¶
- Akceptowane są wyłącznie
mode: sync,text/plainzawarty bezpośrednio w wiadomości icontent_encoding: plain. Dane binarne/base64, zdalne adresy treści, metadane inne niż null, sesje wybrane przez klienta, zadania asynchroniczne, strumieniowanie, odpytywanie, wznawianie i zdalne anulowanie są jawnie odrzucane. Bramka nie pobiera załączników i nie zapisuje wykonań. - Ciała żądań są ograniczone do 65 536 bajtów; liczby wiadomości i części oraz długości tekstu mają osobne limity. Nieobsługiwane pola są odrzucane przed wykonaniem.
- Etykiety ról i poprawne znaczniki czasu SDK są metadanymi transportu. Nazwane role, np.
agent/researcher, są normalizowane doagent; znaczniki czasu są odrzucane. Do kontroli narzędzi trafia tylko tekst i liczbowe oznaczenia ról, więc zakazana litera wtext/plainnie zablokuje przypadkowo niezwiązanej treści. Zadeklarowana rola nigdy nie uwierzytelnia klienta. - Cała odpowiedź agenta jest buforowana w granicach limitu rozmiaru i sprawdzana przed dostarczeniem. Nieudana kontrola wyjścia nie cofa pracy już wykonanej przez agenta. Przekroczenie czasu blokuje wynik i zużywa konserwatywnie zarezerwowane zasoby; zakończenie lokalnego żądania nie gwarantuje zatrzymania pracy zdalnego agenta.
- Bezstanowy klient powinien jawnie przesyłać potrzebny tekst konwersacji w każdym wywołaniu. FastFence nie przechowuje konwersacji ACP ani wyników. Przykładowy backend SDK tworzy własną sesję nawet bez takiego żądania. FastFence sprawdza i odrzuca zwrócony UUID: nigdy nie zwraca, nie zachowuje ani nie używa ponownie identyfikatora sesji. Agent może niezależnie utrzymywać własny stan.
Surowa chroniona reprezentacja narzędzia, dostępna także przez REST/MCP, to {"tool":"acp.assistant","arguments":{"input":[{"role":0,"parts":["Hello"]}]}}. Rola 0 oznacza użytkownika, a 1 agenta. Ta wewnętrzna projekcja tekstu różni się od natywnego schematu ACP. Reguły literalne i semantyczne używają celu Tools.
Oficjalny OpenAPI i klient SDK definiują wiadomości protokołu i zachowanie run_sync używane przez te przykłady.
Kompletne źródła przykładu¶
Agent z oficjalnego SDK¶
"""Actual ACP SDK text agent on loopback, protected by a separate private token."""
import argparse
import hmac
import os
import secrets
from pathlib import Path
import uvicorn
from acp_sdk.models import Message, MessagePart
from acp_sdk.server import Server
from acp_sdk.server.app import create_app
from fastapi import Request
from fastapi.responses import JSONResponse
TOKEN_FILE = Path("state/examples/acp-upstream-token.txt")
server = Server()
@server.agent(
name="uppercase",
input_content_types=["text/plain"],
output_content_types=["text/plain"],
)
async def uppercase(input: list[Message]):
"""Uppercase actual incoming text; no model or simulated business result."""
for message in input:
yield Message(
role="agent/uppercase",
parts=[
MessagePart(
content=part.content.upper(), content_type="text/plain"
)
for part in message.parts
],
)
def private_token() -> str:
TOKEN_FILE.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
try:
descriptor = os.open(
TOKEN_FILE, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600
)
except FileExistsError:
return TOKEN_FILE.read_text().strip()
with os.fdopen(descriptor, "w") as stream:
stream.write(secrets.token_urlsafe(32) + "\n")
return TOKEN_FILE.read_text().strip()
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--port", type=int, default=8020)
args = parser.parse_args()
token = private_token()
if len(token) < 32:
raise SystemExit("Invalid private ACP example credential")
app = create_app(*server.agents, enable_playground_cors=False)
@app.middleware("http")
async def authenticate(request: Request, call_next):
supplied = request.headers.get("authorization", "")
if not hmac.compare_digest(
supplied.encode(), ("Bearer " + token).encode()
):
return JSONResponse(
{"code": "invalid_input", "message": "Authentication required"},
status_code=401,
)
return await call_next(request)
uvicorn.run(app, host="127.0.0.1", port=args.port, access_log=False)
if __name__ == "__main__":
main()
Pobierz acp_server.py · Zobacz źródło
Rejestracja bramki¶
"""Separate example gateway: real ACP forwarding through installed FastFence."""
import argparse
import shutil
from pathlib import Path
import uvicorn
from fastfence.app.factory import create_app
from fastfence.app.interfaces.cli.initialize import initialize
from fastfence.shared.acp import ACPAgentSettings
from fastfence.shared.settings.app_settings import AppSettings
def build_example(upstream_url: str = "http://127.0.0.1:8020"):
token_file = Path("state/examples/acp-upstream-token.txt")
if not token_file.is_file():
raise SystemExit("Start the ACP example server first")
root = Path("state/examples/acp-gateway").resolve()
config = root / "config"
config.mkdir(parents=True, exist_ok=True)
for source, target in (
("acp_policy.yaml", "policy.yaml"),
("signatures.json", "signatures.json"),
):
destination = config / target
if not destination.exists():
shutil.copyfile(Path(__file__).with_name(source), destination)
initialize(root / "state")
return create_app(
AppSettings(
root=root,
state=root / "state",
acp_agents={
"uppercase": ACPAgentSettings(
base_url=upstream_url,
agent_name="uppercase",
api_key=token_file.read_text().strip(),
)
},
)
)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--port", type=int, default=8030)
parser.add_argument("--upstream-url", default="http://127.0.0.1:8020")
args = parser.parse_args()
uvicorn.run(
build_example(args.upstream_url), host="127.0.0.1", port=args.port
)
Pobierz acp_gateway.py · Zobacz źródło
Oficjalny klient SDK¶
"""Call a protected peer agent using the official Agent Communication SDK."""
import argparse
import asyncio
import json
import os
from pathlib import Path
from acp_sdk.client import Client
from acp_sdk.models import Message, MessagePart
async def run(args):
token = os.environ.get("FASTFENCE_AGENT_TOKEN")
if not token:
values = json.loads(args.credentials.read_text())
token = values.get("local-agent") or values["analyst-blue"]
async with Client(
base_url=args.url.rstrip("/") + "/acp",
headers={"Authorization": "Bearer " + token},
timeout=120,
trust_env=False,
) as client:
agents = [agent.name async for agent in client.agents()]
print(json.dumps({"available_agents": agents}))
result = await client.run_sync(
agent=args.agent,
input=[
Message(role="user", parts=[MessagePart(content=args.prompt)])
],
)
print(result.model_dump_json(indent=2))
return result.status.value == "completed"
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--url", default="http://127.0.0.1:8030")
parser.add_argument("--agent", default="uppercase")
parser.add_argument("--prompt", default="hello")
parser.add_argument(
"--credentials",
type=Path,
default=Path("state/examples/acp-gateway/state/credentials.json"),
)
if not asyncio.run(run(parser.parse_args())):
raise SystemExit(1)
if __name__ == "__main__":
main()
Pobierz acp_client.py · Zobacz źródło
Odizolowana polityka¶
version: 1
description: Explicit ACP text-agent example; deterministic checks, no model dependency.
privacy:
enabled: true
input: redact
output: redact
signatures_enabled: true
semantic:
provider: disabled
tools:
acp.uppercase:
roles: [analyst]
timeout_ms: 30000
cost_microusd: 1
models: {}
budgets:
analyst:
calls: 1000
tokens: 1000000
cost_microusd: 1000000
compute_ms: 1000000
concurrent: 2
text_rules:
- id: forbidden-example-input
operator: contains
value: forbidden
direction: input
target: tool