Examples

Runnable recipes, roughly in order of ambition. All are asyncio code; wrap each main() in asyncio.run(main()).

Two nodes, one session

import asyncio
from fg_amp import AgentIdentity, AmpNode, ContactPolicy, InMemoryTransport

async def main():
    inbound = []
    async def on_session(session):
        inbound.append(session)

    alice = AmpNode(identity=AgentIdentity.generate("alice"))
    bob = AmpNode(identity=AgentIdentity.generate("bob"),
                  policy=ContactPolicy.open(), on_session=on_session)

    transport = InMemoryTransport()
    alice.attach(transport)
    bob.attach(transport)

    session = await alice.initiate(bob.card, purpose="price negotiation")
    await session.send_text("Offering 100 units at $4.20 — interested?")
    message = await inbound[0].receive(timeout=1)
    print(message.sender, "→", message.payload.content)
    await session.close()

asyncio.run(main())

Owners, scopes, and a credentialed policy

An owner is the cold root of trust; agents get scoped authority through delegation chains. A credentialed policy that pins trusted_issuers turns scopes into a real gate (without that pin, scopes are advisory — a peer can self-sign any chain).

from fg_amp import AmpNode, ContactPolicy, Delegation, OwnerIdentity

acme = OwnerIdentity.generate("acme-corp")                 # cold root of trust
buyer = AmpNode(identity=acme.create_agent("buyer", {"converse", "negotiate"}))

# The seller only accepts peers whose "negotiate" scope was granted by acme
seller_policy = ContactPolicy.credentialed({"negotiate"}, trusted_issuers={acme.address})

The verified authority surfaces on the session — check the chain, never the messages:

assert session.peer_owner == acme.address
session.require_scope("negotiate", owner=acme.address)   # raises SessionError if not

A chain can also be built explicitly and attached to an identity:

chain = Delegation.grant(principal.keys, principal.address, buyer_id.address,
                         scopes={"converse", "negotiate"}, ttl_seconds=3600)
node = AmpNode(identity=buyer_id.with_delegation(chain))

Groups

A group is a full mesh of pairwise E2E sessions with a founder-signed roster.

group = await analyst.create_group([trader.card, broker.card], purpose="deal room")
await group.send_text("proposal: 500 units at $3.90")      # E2E to every member

# On a member node (register via on_group=collector on the AmpNode):
msg = await member_group.receive(timeout=1)                # GroupMessage
print(msg.sender, msg.payload.content)

await trader_group.leave()                                 # others receive a
event = await member_group.receive(timeout=1)              # GroupEvent
print(event.kind, event.member)                            # "left", trader address

The founder can add_member(card) / remove_member(address); each roster change bumps group.epoch.

Relay + discovery

Run amp-relay --port 8404 on a host, or drive create_relay_app() in-process for tests. Agents register cards, long-poll mailboxes, and discover each other by address — the relay only ever sees ciphertext.

from fg_amp import RelayTransport

relay = RelayTransport("https://relay.example")
await relay.connect(node)                                # register card, start polling

peer_card = await relay.resolve_card("amp:key:…")        # signed-card directory
session = await node.initiate(peer_card, purpose="hello", timeout=10)

In-process variant (as in examples/networked_relay.py): build app = create_relay_app(), drive it with an httpx ASGI client injected as http_call=, then RelayTransport("http://relay.local", http_call=http_call) and await relay_b.connect(bob, poll_interval=0.05).

Wake receiver + non-blocking initiate

For agents that sleep between messages. The recipient advertises a wake URL in its card; the relay POSTs a content-free ping when mail arrives and nobody is polling.

from fg_amp import AmpNode, RelayTransport, WakeNotifier, WakePolicy, WakeReceiver, create_relay_app

# Relay side: enable wake (SSRF-guarded, debounced, non-amplifying)
app = create_relay_app(waker=WakeNotifier(policy=WakePolicy()))

# Recipient side: the wake listener boots the node and pulls
async def on_wake():
    node = AmpNode(identity=me, on_session=handle)
    transport = RelayTransport(relay_url)
    await transport.connect(node)     # pull drains everything waiting

receiver = WakeReceiver(on_wake, path="/wake")
await receiver.start(host="0.0.0.0", port=8080)

# Initiator side: don't block on a sleeping peer
pending = await node.initiate(peer_card, wait=False)
session = await pending.wait(timeout=None)

The wake is a hint — the poll loop is the source of truth, so a dropped ping costs latency, never correctness.

Persistent session + resume

Persistent sessions survive process restarts. The stored record contains no key material; resume re-authenticates, re-verifies the delegation chain fresh, and mints a fresh session key.

from fg_amp import AmpNode, FileSessionStore, SessionMode

node = AmpNode(identity=me, session_store=FileSessionStore("sessions/"))

session = await node.initiate(peer_card, mode=SessionMode.PERSISTENT)
record = node.persist_session(session)          # SessionRecord — safe to store

# ... process restarts ...

session = await node.resume(record.session_id, timeout=30)

MCP bridge

Carry MCP JSON-RPC over an encrypted AMP session (amp.mcp/1). One side exposes a handler, the other calls through it.

from fg_amp import McpBridge

# Server side: expose an MCP handler over the session
async def handler(request: dict) -> dict:
    return await my_mcp_server.dispatch(request)

server_bridge = McpBridge(server_session, handler=handler)
await server_bridge.pump()          # dispatch inbound mcp bodies

# Client side: JSON-RPC call, correlated by id
client_bridge = McpBridge(client_session)
result = await client_bridge.call({"jsonrpc": "2.0", "method": "tools/list"},
                                  timeout=30.0)

Multiple bridges layer on one session — each dispatch returns False for other mcp_session correlators.

Payment lifecycle (x402)

amp.payment/1 carries the x402 objects verbatim; AMP enforces the lifecycle and the sender's delegation-chain spend caps (per-transaction and cumulative, before encrypt).

from fg_amp import PaymentBody, PaymentKind

# Seller quotes
await seller_session.send_body(PaymentBody(
    payment_id="pay-1", kind=PaymentKind.QUOTE,
    amount="4.20", asset="USDC", pay_to="0xSELLER",
    x402={...},                       # x402 payment requirements, verbatim
))

# Buyer authorizes — cap-verified against the buyer's own delegation chain;
# raises SpendRejectedError if over per-tx or cumulative limits
await buyer_session.send_body(PaymentBody(
    payment_id="pay-1", kind=PaymentKind.AUTHORIZATION,
    amount="4.20", asset="USDC", pay_to="0xSELLER",
    x402={...},                       # x402 payment payload
))

# Seller settles out-of-band via x402, then reports
await seller_session.send_body(PaymentBody(
    payment_id="pay-1", kind=PaymentKind.SETTLED, tx_ref="0xabc…",
    amount="4.20", asset="USDC", pay_to="0xSELLER",
))

PaymentTracker refuses illegal transitions (an authorization must reference a known unexpired quote; the quoting side cannot authorize its own quote; only the quoter reports settled/failed) with PaymentLifecycleError. session.spend / session.peer_spend expose the cumulative ledgers.

Transcript verification

Both parties advance an identical SHA-256 hash chain over every envelope. Comparing heads proves an identical, untampered history without exchanging it.

await session.send_text("final terms: 500 units at $3.90")
# ... conversation ends ...

assert session.transcript.head == peer_session.transcript.head
# or against a stored expectation:
assert session.transcript_matches(expected_head)

await session.close("deal reached")   # keys burn; the head survives as evidence