Skip to content

Protect agent-to-agent calls with ACP

FastFence accepts Agent Communication Protocol requests and forwards them to a configured peer agent through the existing tool policy engine:

ACP client → FastFence /acp/runs → input controls → trusted ACP peer
                                                     ↓
ACP response ← output controls ← completed peer response

This is the IBM/BeeAI REST protocol, separate from Agent Client Protocol for editors. Its official repository is archived and the project moved into A2A. FastFence implements a bounded compatibility profile: synchronous, stateless, inline plain-text runs and authenticated agent discovery. It does not claim full ACP or A2A conformance.

Run a real peer agent locally

Install FastFence 1.0.0 or later using Getting started, then extract the complete examples archive into examples/. Run all commands from the installation directory. Keep the gateway in its own uv run FastFence environment. Create a separate environment for the archived SDK peer and client; its Uvicorn pin does not change your gateway. The SDK also imports requests without declaring that dependency, so install it explicitly in this separate environment:

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

The official ACP SDK serves an actual uppercase operation on loopback port 8020. It generates a separate private backend credential in state/examples/acp-upstream-token.txt. All backend endpoints require that credential. The file's contents are never printed.

In a second terminal, run the gateway with its own FastFence dependencies:

uv run --python 3.12 --no-project --with fastfence==1.0.1 python examples/acp_gateway.py

This starts an isolated FastFence instance on port 8030 and registers the uppercase peer as policy tool acp.uppercase. Its configuration and gateway credentials live under state/examples/acp-gateway/. It preserves existing configuration on restart and does not edit your main installation. The explicit example policy uses deterministic checks so no model is needed; the default product policy still requires Laya.

In a third terminal, use the official ACP SDK client:

.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]'

Expected: hello completes with HELLO; forbidden returns a failed run before the peer executes and the script exits 1; the email is redacted before forwarding. HTTP 200 can contain a failed run: check status, error.data.reason and error.data.upstream_executed. Successful output is returned only after output controls pass. Use Activity at http://127.0.0.1:8030 to inspect the decision. The ACP run UUID corresponds to the audit request ID without UUID hyphens.

Server and gateway accept --port; the gateway also accepts --upstream-url. The client accepts --url, --agent and --credentials. Its default credential is the isolated example's local-agent; management credentials cannot invoke ACP.

Connect your existing ACP agent

Configure trusted startup settings in the gateway's environment or private .env:

export FASTFENCE_ACP_AGENTS='{"assistant":{"base_url":"https://peer.example.org","agent_name":"assistant"}}'

Replace the example URL and agent name with your own peer. Add api_key inside that trusted configuration if the peer requires bearer authentication. This is the backend credential, separate from callers' FastFence tokens. It is not returned by discovery or policy APIs. Remote peers require HTTPS; HTTP is accepted only for loopback. Caller input cannot choose upstream URLs or credentials.

Add the corresponding tool to config/policy.yaml, preserving other settings and incrementing the active policy version:

tools:
  acp.assistant:
    roles: [analyst]
    timeout_ms: 30000
    cost_microusd: 1

Restart after changing startup settings. Policy changes still hot-reload. With your normal gateway running on port 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 exposes only registered, configured, role-allowed agents. POST /acp/runs accepts the same bearer identity used by REST/MCP; all these transports share that subject's policy and in-memory budget. Peer costs are the configured tool cost and bounded text resource accounting, not provider billing token measurements. Direct backend access must remain restricted to trusted gateway credentials or network boundaries. These controls cover messages crossing FastFence. They do not inspect a remote agent's internal model/tool calls unless those calls also pass through FastFence; hidden internal token usage is not measured by this adapter.

Supported content and execution

  • Only mode: sync, inline text/plain, and content_encoding: plain are accepted. Binary/base64 parts, remote content URLs, non-null metadata, caller-selected sessions, asynchronous jobs, streaming, polling, resume and remote cancellation are rejected explicitly. The gateway does not fetch attachments or persist runs.
  • Input bodies are bounded to 65,536 bytes; message and part counts and text lengths are bounded separately. Unsupported fields are rejected before execution.
  • Role labels and valid SDK timestamps are transport metadata. Named roles such as agent/researcher normalize to agent; timestamps are discarded. Only text and numeric role markers enter tool controls, so a forbidden letter in text/plain cannot accidentally block unrelated text. Claimed roles never authenticate a caller.
  • The whole peer response is buffered within a size limit and checked before delivery. A failed output check cannot undo work already performed by the peer. Timeouts fail closed and consume conservative reserved resources; ending the local request cannot guarantee that a remote agent stops its own work.
  • Stateless clients should send any necessary conversation text explicitly on each run. FastFence maintains no ACP conversation or result store. The example SDK backend creates its own session even when none is requested. FastFence validates and discards the returned UUID: it never returns, retains or reuses that session identifier. The peer may retain its own state independently of the stateless gateway.

The raw protected tool representation, also available through REST/MCP, is {"tool":"acp.assistant","arguments":{"input":[{"role":0,"parts":["Hello"]}]}}. Role 0 means user and 1 means agent. This internal text projection differs from the native ACP wire schema. Literal and semantic rules use target Tools.

The official OpenAPI and SDK client define the wire messages and run_sync behavior used by these examples.

Complete example sources

Official SDK peer

"""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()

Download acp_server.py · View source

Gateway registration

"""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
    )

Download acp_gateway.py · View source

Official SDK client

"""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()

Download acp_client.py · View source

Isolated policy

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

Download acp_policy.yaml · View source