diff --git a/extensions/settlement/README.md b/extensions/settlement/README.md new file mode 100644 index 000000000..2e81c84a2 --- /dev/null +++ b/extensions/settlement/README.md @@ -0,0 +1,27 @@ +# Settlement Extension + +This directory contains the specification for the A2A Settlement Extension +(A2A-SE), which adds escrow-based token settlement to the A2A task lifecycle. + +The extension enables agents to: + +- Declare skill-level pricing in their Agent Cards +- Hold funds in escrow while work is in progress +- Release payment on task completion, refund on failure +- Resolve disputes when requester and provider disagree + +A2A-SE is a **data + profile extension**: it adds structured settlement data to +Agent Cards and overlays settlement metadata onto the core request-response +messages via the `metadata` field. It does not add new RPC methods or task +states to the core protocol. + +The v1 directory contains the specification document. A library implementation +in Python is present in `samples/python/extensions/settlement`. A live reference +exchange is available at . + +## Resources + +- [Full specification](https://github.com/a2a-settlement/a2a-settlement/blob/main/SPEC.md) +- Python SDK: `pip install a2a-settlement` ([source](https://github.com/a2a-settlement/a2a-settlement/tree/main/sdk)) +- TypeScript SDK: `npm install @a2a-settlement/sdk` ([source](https://github.com/a2a-settlement/a2a-settlement/tree/main/sdk-ts)) +- [Live exchange API docs](https://exchange.a2a-settlement.org/docs) diff --git a/extensions/settlement/v1/spec.md b/extensions/settlement/v1/spec.md new file mode 100644 index 000000000..5d022d5e4 --- /dev/null +++ b/extensions/settlement/v1/spec.md @@ -0,0 +1,265 @@ +# A2A Settlement Extension (A2A-SE) — v1 + +## Overview + +The A2A Settlement Extension adds escrow-based token settlement to the A2A task +lifecycle. Agents declare pricing in their Agent Cards. Clients create escrow +before sending a task, and release or refund based on the terminal task state. + +A2A-SE requires zero modifications to the core A2A protocol. It uses the +existing `capabilities.extensions` mechanism for Agent Card integration, and +the existing `metadata` field on Messages and Tasks for settlement context. + +## Extension URI + +```text +https://a2a-settlement.org/extensions/settlement/v1 +``` + +## Agent Card Declaration + +Agents that support settlement declare the extension in their Agent Card's +`capabilities.extensions` array: + +```json +{ + "capabilities": { + "extensions": [ + { + "uri": "https://a2a-settlement.org/extensions/settlement/v1", + "description": "Accepts token-based payment via A2A Settlement Exchange", + "required": false, + "params": { + "exchangeUrls": [ + "https://exchange.a2a-settlement.org/api/v1" + ], + "preferredExchange": "https://exchange.a2a-settlement.org/api/v1", + "accountIds": { + "https://exchange.a2a-settlement.org/api/v1": "provider-uuid" + }, + "pricing": { + "sentiment-analysis": { + "baseTokens": 10, + "model": "per-request", + "currency": "ATE" + } + }, + "reputation": 0.87, + "availability": 0.95 + } + } + ] + } +} +``` + +### Extension Params + +| Field | Type | Required | Description | +|---------------------|----------|----------|------------------------------------------------------------| +| `exchangeUrls` | string[] | Yes | Exchange endpoints the agent is registered on | +| `preferredExchange` | string | No | The agent's preferred exchange from the list | +| `accountIds` | object | Yes | Map of exchange URL to agent's account ID on that exchange | +| `pricing` | object | No | Map of skill ID to pricing configuration | +| `currency` | string | No | Default currency (default: `ATE`) | +| `reputation` | number | No | Agent's reputation score (0.0 – 1.0) | +| `availability` | number | No | Agent's availability score (0.0 – 1.0) | + +When `required` is `false`, the agent accepts both paid and unpaid requests +(freemium model). When `required` is `true`, the agent rejects tasks that do +not include settlement metadata. + +### Pricing Models + +| Model | Description | Example | +|---------------|------------------------------------------|----------------------------------| +| `per-request` | Fixed cost per task invocation | 10 tokens per sentiment analysis | +| `per-unit` | Cost per unit of input (per 1K chars) | 2 tokens per 1,000 characters | +| `per-minute` | Cost per minute of processing time | 5 tokens per minute of compute | +| `negotiable` | Price determined during task negotiation | Agent proposes price in response | + +## Settlement Flow Mapped to A2A TaskStates + +Settlement actions map directly to existing A2A TaskState transitions. No new +task states are required. + +```text +A2A TaskState Settlement Action +───────────── ───────────────── +SUBMITTED ──────► Client creates escrow on exchange +WORKING ──────► No action (escrow holds) +INPUT_REQUIRED ──────► No action (escrow holds during multi-turn) +COMPLETED ──────► Client releases escrow (tokens → provider) +FAILED ──────► Client refunds escrow (tokens → client) +CANCELED ──────► Client refunds escrow (tokens → client) +REJECTED ──────► Client refunds escrow (tokens → client) +``` + +## Settlement Metadata + +Settlement context is passed through A2A's existing `metadata` field using a +namespaced key `a2a-se` to avoid collisions. + +### Client's Initial Message + +When creating a task, the client includes the escrow reference: + +```json +{ + "messageId": "msg-uuid", + "role": "user", + "parts": [ + { "text": "Analyze the sentiment of this earnings transcript." } + ], + "metadata": { + "a2a-se": { + "escrowId": "escrow-uuid-from-exchange", + "amount": 10, + "feeAmount": 1, + "exchangeUrl": "https://exchange.a2a-settlement.org/api/v1", + "expiresAt": "2026-02-17T12:30:00Z" + } + } +} +``` + +### Provider's Task Response + +The provider acknowledges the escrow in its response metadata: + +```json +{ + "id": "task-uuid", + "status": { + "state": "TASK_STATE_WORKING" + }, + "metadata": { + "a2a-se": { + "escrowId": "escrow-uuid-from-exchange", + "settlementStatus": "acknowledged" + } + } +} +``` + +### Settlement Status Values + +| Status | Meaning | +|----------------|----------------------------------------------------| +| `pending` | Escrow created, awaiting agent acknowledgment | +| `acknowledged` | Agent confirmed receipt of escrow reference | +| `review` | Task completed, requester reviewing before release | +| `released` | Tokens transferred to provider | +| `refunded` | Tokens returned to requester | +| `expired` | Escrow TTL exceeded without resolution | +| `disputed` | Transaction flagged by either party | + +## Settlement Exchange API + +The settlement exchange is an external REST API that manages accounts, escrow, +and token balances. The exchange is an **interface**: any conforming +implementation (hosted, self-hosted, on-chain) can serve as the settlement +rail. + +### Core Endpoints + +| Method | Path | Description | +|--------|--------------------------|-----------------------------------------------------------------------| +| `POST` | `/exchange/escrow` | Create an escrow | +| `POST` | `/exchange/release` | Release escrowed tokens to provider. Body: `{"escrow_id": ""}` | +| `POST` | `/exchange/refund` | Refund escrowed tokens to requester. Body: `{"escrow_id": ""}` | +| `POST` | `/exchange/dispute` | Flag an escrow as disputed. Body: `{"escrow_id": ""}` | +| `GET` | `/exchange/escrows/{id}` | Look up an escrow by ID | +| `GET` | `/exchange/balance` | Get agent's token balance | + +### Batch Escrow for Pipelines + +Multi-step workflows can create all escrows atomically with dependency ordering: + +```json +{ + "group_id": "pipeline-001", + "escrows": [ + { + "provider_id": "translator-agent", + "amount": 10, + "task_type": "translate" + }, + { + "provider_id": "sentiment-agent", + "amount": 15, + "task_type": "sentiment", + "depends_on": ["$0"] + } + ] +} +``` + +Values in `depends_on` use positional references (`$0`, `$1`, ...) referring to +other escrows in the same batch by their zero-based index. The field models +sequential dependencies: an escrow cannot be released until all upstream escrows +have been released. When an upstream escrow is refunded, all downstream escrows +are automatically cascade-refunded. + +## Extension Activation + +Clients activate the settlement extension using the `A2A-Extensions` HTTP +header: + +```http +POST /agents/provider HTTP/1.1 +A2A-Extensions: https://a2a-settlement.org/extensions/settlement/v1 +Content-Type: application/json + +{ + "jsonrpc": "2.0", + "method": "message/send", + "id": "1", + "params": { + "message": { + "messageId": "msg-1", + "role": "user", + "parts": [{"text": "Analyze this text"}], + "metadata": { + "a2a-se": { + "escrowId": "esc-uuid", + "amount": 10, + "exchangeUrl": "https://exchange.a2a-settlement.org/api/v1" + } + } + } + } +} +``` + +## Client Workflow + +1. **Discover** provider via Agent Card; check for settlement extension URI. +2. **Negotiate exchange** — intersect `exchangeUrls` with client's exchanges. +3. **Create escrow** on the selected exchange. +4. **Send A2A message** with `a2a-se` metadata containing the `escrowId`. +5. **On terminal state** — release (COMPLETED) or refund (FAILED/CANCELED/REJECTED). + +## Provider Workflow + +1. **Declare** settlement extension in Agent Card with pricing. +2. **On incoming message** — read `metadata["a2a-se"]["escrowId"]`. +3. **Verify escrow** by calling `GET /exchange/escrows/{id}` on the exchange. +4. **Execute task** normally via A2A. +5. Requester handles release/refund based on task outcome. + +## Security Considerations + +- All exchange endpoints require Bearer token authentication. +- Escrow creation supports idempotency keys to prevent duplicate holds. +- Escrows have configurable TTL (default 30 minutes) with automatic expiration. +- Disputes freeze the escrow until operator resolution. +- The extension MUST NOT bypass the agent's primary security controls. + +## Reference Implementation + +- Specification: +- Python SDK: `pip install a2a-settlement` +- TypeScript SDK: `npm install @a2a-settlement/sdk` +- Live exchange: +- Live stats: diff --git a/samples/python/extensions/settlement/README.md b/samples/python/extensions/settlement/README.md new file mode 100644 index 000000000..4267cd336 --- /dev/null +++ b/samples/python/extensions/settlement/README.md @@ -0,0 +1,44 @@ +# Settlement Extension Implementation + +This is the Python implementation of the A2A Settlement Extension defined in +`extensions/settlement/v1`. + +## What it does + +Adds escrow-based token settlement to A2A agent interactions. Funds are held in +escrow while an agent works on a task, then released on completion or refunded +on failure. + +## Quick start + +```python +from settlement_ext import SettlementExtension + +ext = SettlementExtension( + exchange_url="https://exchange.a2a-settlement.org/api/v1", + api_key="ate_your_key", + account_id="your-agent-uuid", +) + +# Add to your AgentCard +card = ext.add_to_card(card, pricing={ + "sentiment-analysis": {"baseTokens": 10, "model": "per-request"} +}) + +# Server side: wrap your executor to auto-verify escrow +agent_executor = ext.wrap_executor(agent_executor) + +# Client side: wrap your client to auto-settle on task completion +client = ext.wrap_client(client) +``` + +## Usage patterns + +The extension provides several integration levels, from fully manual to fully +managed. See the `SettlementExtension` class documentation for all options. + +## Live exchange + +A public sandbox exchange is available at + for testing. Register an agent to +get an API key and 100 starter tokens. diff --git a/samples/python/extensions/settlement/pyproject.toml b/samples/python/extensions/settlement/pyproject.toml new file mode 100644 index 000000000..9a446554f --- /dev/null +++ b/samples/python/extensions/settlement/pyproject.toml @@ -0,0 +1,11 @@ +[project] +name = "settlement_ext" +version = "0.1.0" +description = "A2A Settlement Extension — escrow-based payment for A2A agents" +readme = "README.md" +requires-python = ">=3.10" +dependencies = ["a2a-sdk>=0.3.0", "a2a-settlement>=0.8.0"] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" diff --git a/samples/python/extensions/settlement/src/settlement_ext/__init__.py b/samples/python/extensions/settlement/src/settlement_ext/__init__.py new file mode 100644 index 000000000..36656f138 --- /dev/null +++ b/samples/python/extensions/settlement/src/settlement_ext/__init__.py @@ -0,0 +1,570 @@ +"""A2A Settlement Extension — escrow-based payment for A2A agents. + +Provides SettlementExtension with multiple integration levels: + + Option 1 (manual): use the helpers to build metadata and call the SDK yourself. + Option 2 (server-side wrap): wrap_executor auto-verifies escrow before execution. + Option 3 (client-side wrap): wrap_client auto-creates escrow before sending + and auto-settles (release/refund) when the task reaches a terminal state. +""" + +from __future__ import annotations + +import asyncio +import logging +import re +import time + +from typing import TYPE_CHECKING, Any + +from a2a.client import Client, ClientCallInterceptor, ClientEvent +from a2a.extensions.common import find_extension_by_uri +from a2a.server.agent_execution import AgentExecutor, RequestContext +from a2a.types import ( + AgentCard, + AgentExtension, + GetTaskPushNotificationConfigParams, + Message, + Task, + TaskIdParams, + TaskPushNotificationConfig, + TaskQueryParams, + TaskState, + TaskStatusUpdateEvent, +) +from a2a_settlement import SettlementExchangeClient +from a2a_settlement.agentcard import build_settlement_extension +from a2a_settlement.lifecycle import settle_for_task_state +from a2a_settlement.metadata import ( + build_settlement_metadata, + get_settlement_block, +) + + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + + from a2a.client.middleware import ClientCallContext + from a2a.server.events.event_queue import EventQueue + +logger = logging.getLogger(__name__) + +URI = 'https://a2a-settlement.org/extensions/settlement/v1' +METADATA_KEY = 'a2a-se' + +_TERMINAL_RELEASE = {TaskState.completed, 'TASK_STATE_COMPLETED'} +_TERMINAL_REFUND = { + TaskState.failed, + TaskState.canceled, + TaskState.rejected, + 'TASK_STATE_FAILED', + 'TASK_STATE_CANCELED', + 'TASK_STATE_REJECTED', +} +_MESSAGING_METHODS = {'message/send', 'message/stream'} + +_ESCROW_ID_RE = re.compile(r'^[A-Za-z0-9_\-]{1,128}$') + + +def _valid_escrow_id(value: Any) -> bool: + """Return True if *value* looks like a well-formed escrow ID.""" + return isinstance(value, str) and bool(_ESCROW_ID_RE.fullmatch(value)) + + +class SettlementExtension: + """A2A Settlement Extension implementation. + + Mirrors the pattern of the timestamp extension, offering multiple + integration levels from fully manual to fully managed. + + Attributes: + auto_verify: If True, the executor wrapper rejects tasks whose + escrow cannot be verified on the exchange. + auto_settle: If True, the client wrapper releases/refunds escrow + automatically when a task reaches a terminal state. + **Security note**: when enabled, the provider controls when + payment is released by setting the task to COMPLETED. Set to + False and call ``release()`` explicitly when the client must + review the outcome before payment. + settlement_required: If True, the executor wrapper rejects tasks + that activate settlement but do not include settlement + metadata. Default False (freemium — unpaid requests allowed). + + Args: + exchange_url: Base URL of the settlement exchange. + api_key: Bearer token for the exchange. + account_id: This agent's account ID on the exchange. + auto_verify: If True, the executor wrapper rejects tasks whose + escrow cannot be verified on the exchange. Default True. + auto_settle: If True, the client wrapper releases/refunds escrow + automatically when a task reaches a terminal state. + Default False. + """ + + def __init__( + self, + exchange_url: str, + api_key: str, + account_id: str, + *, + auto_verify: bool = True, + auto_settle: bool = False, + ) -> None: + self._exchange_url = exchange_url + self._api_key = api_key + self.account_id = account_id + self.auto_verify = auto_verify + self.auto_settle = auto_settle + self.settlement_required = False + self.exchange_client = SettlementExchangeClient( + base_url=exchange_url, + api_key=api_key, + ) + + # ── AgentCard integration ────────────────────────────────── + + def agent_extension( + self, + pricing: dict[str, Any] | None = None, + *, + required: bool = False, + ) -> AgentExtension: + """Build the AgentExtension object for this extension.""" + ext_dict = build_settlement_extension( + exchange_urls=[self._exchange_url], + account_ids={self._exchange_url: self.account_id}, + pricing=pricing, + required=required, + ) + return AgentExtension(**ext_dict) + + def add_to_card( + self, + card: AgentCard, + pricing: dict[str, Any] | None = None, + *, + required: bool = False, + ) -> AgentCard: + """Add the settlement extension to an AgentCard.""" + if not card.capabilities.extensions: + card.capabilities.extensions = [] + card.capabilities.extensions.append( + self.agent_extension(pricing, required=required) + ) + return card + + def is_supported(self, card: AgentCard | None) -> bool: + """Check whether an AgentCard advertises settlement.""" + if card: + return find_extension_by_uri(card, URI) is not None + return False + + # ── Extension activation ─────────────────────────────────── + + def activate(self, context: RequestContext) -> bool: + """Activate the extension if the client requested it.""" + if URI in context.requested_extensions: + context.add_activated_extension(URI) + return True + return False + + # ── Manual helpers (Option 1) ────────────────────────────── + + def create_escrow( + self, + *, + provider_id: str, + amount: int, + task_id: str | None = None, + task_type: str | None = None, + ttl_minutes: int | None = None, + ) -> dict[str, Any]: + """Create an escrow on the exchange.""" + return self.exchange_client.create_escrow( + provider_id=provider_id, + amount=amount, + task_id=task_id, + task_type=task_type, + ttl_minutes=ttl_minutes, + ) + + def release(self, escrow_id: str) -> dict[str, Any]: + """Release an escrow (pay the provider).""" + return self.exchange_client.release_escrow(escrow_id=escrow_id) + + def refund( + self, + escrow_id: str, + reason: str | None = None, + ) -> dict[str, Any]: + """Refund an escrow (return tokens to requester).""" + return self.exchange_client.refund_escrow( + escrow_id=escrow_id, reason=reason + ) + + def verify_escrow(self, escrow_id: str) -> dict[str, Any] | None: + """Look up an escrow on the exchange. + + Returns: + The escrow dict, or None if the lookup fails. + """ + try: + return self.exchange_client.get_escrow(escrow_id=escrow_id) + except (OSError, ValueError, KeyError): + logger.warning( + 'Failed to verify escrow %s', + escrow_id, + exc_info=True, + ) + return None + + def build_metadata(self, escrow: dict[str, Any]) -> dict[str, Any]: + """Build the a2a-se metadata block from an escrow.""" + meta = build_settlement_metadata( + escrow_id=escrow['escrow_id'], + amount=escrow['amount'], + fee_amount=escrow.get('fee_amount', 0), + exchange_url=self._exchange_url, + expires_at=escrow.get('expires_at', ''), + ) + if not escrow.get('expires_at'): + meta['a2a-se'].pop('expiresAt', None) + return meta + + @staticmethod + def read_metadata( + message: Message | Task | dict, + ) -> dict[str, Any] | None: + """Extract the a2a-se block from a message or task.""" + return get_settlement_block(message) + + # ── Server-side: executor wrapper (Option 2) ─────────────── + + def wrap_executor(self, executor: AgentExecutor) -> AgentExecutor: + """Wrap an executor to verify escrow before execution. + + If the settlement extension is activated and the incoming + message contains an escrowId, the wrapper verifies the + escrow exists and is in 'held' status on the exchange + before delegating to the real executor. If verification + fails, the task is rejected. + """ + return _SettledAgentExecutor(executor, self) + + # ── Client-side: client wrapper (Option 3) ───────────────── + + def wrap_client(self, client: Client) -> Client: + """Wrap a client to auto-manage escrow lifecycle. + + Outgoing messages get escrow metadata injected. When the + task reaches a terminal state, escrow is released or + refunded. + """ + return _SettledClient(client, self) + + def client_interceptor(self) -> ClientCallInterceptor: + """Get a client interceptor that activates settlement.""" + return _SettlementClientInterceptor(self) + + +# ── Server-side wrapper ──────────────────────────────────────── + + +class _SettledAgentExecutor(AgentExecutor): + """Verifies escrow before delegating to the wrapped executor.""" + + def __init__( + self, + delegate: AgentExecutor, + ext: SettlementExtension, + ) -> None: + self._delegate = delegate + self._ext = ext + self._used_escrows: dict[str, float] = {} + self._verify_lock = asyncio.Lock() + + async def execute( + self, + context: RequestContext, + event_queue: EventQueue, + ) -> None: + if self._ext.activate(context) and self._ext.auto_verify: + se_block = self._extract_settlement(context) + reject = False + if se_block and not await self._verify(se_block): + reject = True + elif not se_block and self._ext.settlement_required: + logger.warning('Settlement required but no metadata provided') + reject = True + + if reject: + await event_queue.enqueue_event( + TaskStatusUpdateEvent( + taskId=context.task_id or '', + contextId=context.context_id or '', + final=True, + status=TaskStatusUpdateEvent.Status( + state=TaskState.failed, + message=Message( + messageId='settlement-reject', + role='agent', + parts=[], + ), + ), + ) + ) + return + await self._delegate.execute(context, event_queue) + + async def cancel( + self, + context: RequestContext, + event_queue: EventQueue, + ) -> None: + await self._delegate.cancel(context, event_queue) + + def _extract_settlement(self, context: RequestContext) -> dict | None: + msg = context.message + if msg and msg.metadata: + return msg.metadata.get(METADATA_KEY) + return None + + async def _verify(self, se_block: dict) -> bool: + escrow_id = se_block.get('escrowId') + if not _valid_escrow_id(escrow_id): + logger.warning('Invalid or missing escrow ID') + return False + + async with self._verify_lock: + self._prune_used() + if escrow_id in self._used_escrows: + logger.warning('Escrow already used') + return False + self._used_escrows[escrow_id] = time.monotonic() + + escrow = await asyncio.to_thread(self._ext.verify_escrow, escrow_id) + if not escrow: + return False + + reason = self._check_escrow(se_block, escrow, escrow_id) + if reason: + logger.warning('Escrow %s rejected: %s', escrow_id, reason) + return False + return True + + def _prune_used(self) -> None: + cutoff = time.monotonic() - _ESCROW_TTL_S + stale = [k for k, ts in self._used_escrows.items() if ts < cutoff] + for k in stale: + self._used_escrows.pop(k, None) + + def _check_escrow( + self, se_block: dict, escrow: dict, escrow_id: str + ) -> str | None: + if escrow.get('provider_id') != self._ext.account_id: + return ( + f'provider mismatch: expected {self._ext.account_id}, ' + f'got {escrow.get("provider_id")}' + ) + expected_amount = se_block.get('amount') + if ( + expected_amount is not None + and escrow.get('amount') != expected_amount + ): + return ( + f'amount mismatch: expected {expected_amount}, ' + f'got {escrow.get("amount")}' + ) + if escrow.get('status') != 'held': + return f'unexpected status: {escrow.get("status")}' + return None + + +# ── Client-side wrapper ──────────────────────────────────────── + + +_ESCROW_TTL_S = 30 * 60 # 30 minutes, matches default exchange escrow TTL + + +class _SettledClient(Client): + """Manages escrow around task messages. + + Keeps a mapping of task_id to escrow_id so it can settle + automatically when the task reaches a terminal state. Entries + are pruned after ``_ESCROW_TTL_S`` to bound memory growth. + """ + + def __init__(self, delegate: Client, ext: SettlementExtension) -> None: + self._delegate = delegate + self._ext = ext + self._escrows: dict[str, tuple[str, float]] = {} + + async def send_message( + self, + request: Message, + *, + context: ClientCallContext | None = None, + ) -> AsyncIterator[ClientEvent | Message]: + se_block = self._ext.read_metadata(request) + raw_id = se_block.get('escrowId') if se_block else None + escrow_id = raw_id if _valid_escrow_id(raw_id) else None + tracked = False + + async for event in self._delegate.send_message( + request, context=context + ): + yield event + if escrow_id and not tracked: + task_id, _ = self._extract_task_state(event) + if task_id: + self.track_escrow(task_id, escrow_id) + tracked = True + if self._ext.auto_settle: + await self._try_settle(event) + + async def get_task( + self, + request: TaskQueryParams, + *, + context: ClientCallContext | None = None, + ) -> Task: + task = await self._delegate.get_task(request, context=context) + if self._ext.auto_settle: + await self._try_settle(task) + return task + + async def cancel_task( + self, + request: TaskIdParams, + *, + context: ClientCallContext | None = None, + ) -> Task: + task = await self._delegate.cancel_task(request, context=context) + if self._ext.auto_settle: + await self._try_settle(task) + return task + + async def set_task_callback( + self, + request: TaskPushNotificationConfig, + *, + context: ClientCallContext | None = None, + ) -> TaskPushNotificationConfig: + return await self._delegate.set_task_callback(request, context=context) + + async def get_task_callback( + self, + request: GetTaskPushNotificationConfigParams, + *, + context: ClientCallContext | None = None, + ) -> TaskPushNotificationConfig: + return await self._delegate.get_task_callback(request, context=context) + + async def resubscribe( + self, + request: TaskIdParams, + *, + context: ClientCallContext | None = None, + ) -> AsyncIterator[ClientEvent]: + async for event in self._delegate.resubscribe(request, context=context): + yield event + if self._ext.auto_settle: + await self._try_settle(event) + + async def get_card( + self, + *, + context: ClientCallContext | None = None, + ) -> AgentCard: + return await self._delegate.get_card(context=context) + + def track_escrow(self, task_id: str, escrow_id: str) -> None: + """Register an escrow for auto-settlement.""" + self._prune_stale() + self._escrows[task_id] = (escrow_id, time.monotonic()) + + def _prune_stale(self) -> None: + cutoff = time.monotonic() - _ESCROW_TTL_S + stale = [k for k, (_, ts) in self._escrows.items() if ts < cutoff] + for k in stale: + self._escrows.pop(k, None) + + async def _try_settle(self, event: Any) -> None: + task_id, state = self._extract_task_state(event) + if not task_id or not state: + return + + entry = self._escrows.get(task_id) + if not entry: + return + escrow_id = entry[0] + + try: + await asyncio.to_thread( + settle_for_task_state, + self._ext.exchange_client, + task_state=state, + escrow_id=escrow_id, + ) + except (OSError, ValueError, KeyError): + logger.warning( + 'Auto-settle failed for escrow %s', + escrow_id, + exc_info=True, + ) + return + + if state in _TERMINAL_RELEASE | _TERMINAL_REFUND: + self._escrows.pop(task_id, None) + + @staticmethod + def _extract_task_state( + event: Any, + ) -> tuple[str | None, str | None]: + if isinstance(event, Task): + return ( + event.id, + event.status.state if event.status else None, + ) + if isinstance(event, TaskStatusUpdateEvent): + return ( + event.taskId, + event.status.state if event.status else None, + ) + return None, None + + +class _SettlementClientInterceptor(ClientCallInterceptor): + """Activates the settlement extension on outgoing requests.""" + + def __init__(self, ext: SettlementExtension) -> None: + self._ext = ext + + async def intercept( + self, + method_name: str, + request_payload: dict[str, Any], + http_kwargs: dict[str, Any], + agent_card: AgentCard | None, + context: ClientCallContext | None, + ) -> tuple[dict[str, Any], dict[str, Any]]: + if ( + not self._ext.is_supported(agent_card) + or method_name not in _MESSAGING_METHODS + ): + return (request_payload, http_kwargs) + + headers = http_kwargs.setdefault('headers', {}) + header_key = 'A2A-Extensions' + existing = headers.get(header_key, '') + if URI not in existing: + headers[header_key] = f'{existing}, {URI}'.lstrip(', ') + + return (request_payload, http_kwargs) + + +__all__ = [ + 'METADATA_KEY', + 'URI', + 'SettlementExtension', +] diff --git a/samples/python/extensions/settlement/src/settlement_ext/py.typed b/samples/python/extensions/settlement/src/settlement_ext/py.typed new file mode 100644 index 000000000..e69de29bb