--- name: fet-payment-protocol description: Self-contained recipe for adding native FET (Fetch.ai mainnet / stable-testnet) on-chain payments to a uAgent alongside Chat Protocol. Includes copy-paste-ready code for the FET ledger verifier, payment protocol handlers, chat protocol that triggers payment, agent entrypoint, and `.env`. Covers `Funds(payment_method="fet_direct")`, `RequestPayment` metadata (`fet_network`, `provider_agent_wallet`), `CommitPayment` verification via `cosmpy`, idempotency on `transaction_id`, `RejectPayment`, and `CompletePayment`. Use when adding direct FET payments to a Fetch.ai uAgent, gating chat actions behind on-chain FET transfer, or wiring `payment_protocol_spec` next to `chat_protocol_spec` for a crypto-only flow. --- # FET Payment Protocol (Fetch.ai uAgents) ## Purpose Give the coding agent everything it needs — file layout, full code, env vars — to add a working **direct FET payment protocol** to a Fetch.ai uAgent that also speaks Chat Protocol. The buyer pays the seller agent's wallet on-chain (mainnet `afet` or stable-testnet `atestfet`); the agent verifies the transfer against the ledger before fulfilling. No Stripe, no card processor, no off-chain escrow. ## When to Use - Adding native FET payments to a chat-capable uAgent - Implementing `uagents_core.contrib.protocols.payment` with `payment_method="fet_direct"` - Gating a chat action (image gen, video gen, search, paid API, etc.) behind an on-chain FET transfer - Any task referencing `RequestPayment`, `CommitPayment`, `CompletePayment`, `RejectPayment`, `CancelPayment`, or `payment_protocol_spec` together with FET / `fet_direct` / `cosmpy` ## When NOT to Use - Card / Stripe payments — use the `stripe-payment-protocol` skill instead - Smart-contract escrow or staking flows — this skill targets a single direct transfer - Mock / pseudo payment flows - Payments in any non-FET denom on Fetch.ai (USDC, etc.) --- ## Package Setup Before running install commands, creating `pyproject.toml`, or generating `.env.example` / `.gitignore`, check which **package setup skill** is installed and defer to it: - `uv-package` for `uv` projects - `poetry-package` for Poetry projects - `python-venv-package` for `venv` + `pip` projects - `no-package` for code-only mode (no install commands) This protocol skill only defines the FET payment protocol and ledger verification code. It does not own dependency installation, lockfile management, or virtualenv creation. ### Package Skill Precedence (mandatory) Use this decision order before emitting any setup/install/run command: 1. Detect which package skills are actually available in the current session from this set only: `uv-package`, `no-package`, `poetry-package`, `python-venv-package`. 2. If the user explicitly asks for one of those managers and the matching skill is available, use it. 3. Otherwise, if repository signals are present, select the matching available skill: - `uv.lock` or `[tool.uv]` -> `uv-package` - `poetry.lock` or `[tool.poetry]` -> `poetry-package` - `requirements.txt` / existing `.venv` conventions -> `python-venv-package` 4. If no manager is detectable, choose one available skill and stay consistent for the full response. Prefer this fallback order: `uv-package` -> `poetry-package` -> `python-venv-package` -> `no-package`. 5. Never emit commands for a package skill that is not available. When a package skill is active, all dependency/setup/run instructions must come from that skill only: - `uv-package` -> `uv add ...`, `uv run python agent.py` - `poetry-package` -> `poetry add ...`, `poetry run python agent.py` - `python-venv-package` -> `pip install ...` (and project's venv flow) - `no-package` -> no install commands; only document dependencies Extra runtime dependencies required by this skill on top of the base set: `cosmpy` (Fetch.ai ledger client) and `openai` (ASI:One default LLM client). Add them through whichever package skill is active: | Active package skill | Command | | --- | --- | | `uv-package` | `uv add cosmpy openai` | | `poetry-package` | `poetry add cosmpy openai` | | `python-venv-package` | `pip install cosmpy openai && pip freeze > requirements.txt` | | `no-package` | document `cosmpy` and `openai` in the project's `## Dependencies` block | --- ## File Layout Build this structure exactly. Every file has a single concern: ``` agent.py # entrypoint: load env, create Agent, set wallet, include both protocols protocols/ __init__.py # empty chat_proto.py # chat protocol + handler that triggers payment when needed payment_proto.py # payment protocol + handlers + request_payment_from_user helper fet_payments/ __init__.py # empty ledger.py # ONLY place that touches cosmpy / the Fetch.ai ledger .env.example ``` Hard separation — never mix: - `cosmpy` calls and ledger queries live **only** in `fet_payments/ledger.py`. - `RequestPayment` / `CommitPayment` / `RejectPayment` / `CompletePayment` / `CancelPayment` handling lives **only** in `protocols/payment_proto.py`. - `ChatMessage` handling lives **only** in `protocols/chat_proto.py`. - `agent.py` only loads env, creates the `Agent`, calls `set_agent_wallet(agent.wallet)`, and `include`s both protocols. --- ## The Payment Flow ``` ChatMessage arrives protocols/chat_proto.py: handle_chat └─ if user already has verified_payment for this session -> fulfill paid action └─ if request needs payment -> call request_payment_from_user(...) request_payment_from_user (in protocols/payment_proto.py) └─ persist user's prompt under "chat_prompt::" └─ build accepted_funds=[Funds(currency="FET", amount=FET_AMOUNT_FET, payment_method="fet_direct")] └─ build metadata={"fet_network": "stable-testnet"|"mainnet", "mainnet": "false"|"true", "provider_agent_wallet": str(agent.wallet.address())} └─ ctx.send(user_address, RequestPayment(...)) └─ ctx.storage.set("payment_request:pending::", ...) User signs + broadcasts an on-chain FET transfer to provider_agent_wallet. The user-side client sends back: CommitPayment(transaction_id=, funds=Funds(currency="FET", amount="...", payment_method="fet_direct"), metadata={"buyer_fet_wallet": "fetch1..."}) protocols/payment_proto.py: on_commit ├─ idempotency check: │ already processed? │ ├─ fulfilled already -> re-send CompletePayment │ └─ not fulfilled yet -> re-send CompletePayment + resume run_paid_action(...) ├─ extract buyer wallet from msg.metadata (buyer_fet_wallet | buyer_fet_address) ├─ asyncio.to_thread(verify_fet_payment_to_agent, │ transaction_id=msg.transaction_id, │ expected_amount_fet=str(msg.funds.amount), │ sender_fet_address=buyer_fet_wallet, │ recipient_agent_wallet=_agent_wallet, │ use_mainnet=not use_testnet) ├─ verified -> resolve chat session id (latest pending session fallback -> ctx.session) │ -> mark processed + verified_payment │ -> ctx.send(sender, CompletePayment(transaction_id=...)) │ -> read storage["chat_prompt::"] │ -> await run_paid_action(ctx, sender, session_id, text=prompt) │ -> mark fulfillment complete └─ not verified / missing wallet / disabled -> ctx.send(sender, CancelPayment(reason=...)) If buyer/UI sends RejectPayment -> on_reject clears pending state and notifies via chat. ``` The seller agent **never** holds buyer keys. It only verifies that a tx with the expected amount, sender, recipient, and denom landed on the chosen network. --- ## Directional Logging Contract Every agent generated from this skill must emit structured logs so the user can trace the FET payment lifecycle end-to-end from the CLI. This makes on-chain debugging tractable and keeps multi-agent interactions auditable. ### Required tags - `[inbound]` — log on entry to every protocol handler (`ChatMessage`, `ChatAcknowledgement`, `CommitPayment`, `RejectPayment`), **before** any processing. Include message type + sender. - `[outbound]` — log immediately **before** every `ctx.send(...)`. Include message type + recipient. - `[state]` — log notable storage transitions (pending request saved/cleared, `verified_payment` set/cleared, duplicate short-circuits). - `[payment]` — domain-specific FET events (ledger query start, verify result, per-network attempt). ### Redaction rules - Never log `AGENT_SEED`, raw private keys, or full signed transactions. - Log only `tx_id`, `buyer_fet_wallet`, `recipient_agent_wallet`, amount in FET + denom, and network (`mainnet` / `stable-testnet`). - Truncate user-supplied description / chat text at ~120 chars. ### Helpers (put in `protocols/payment_proto.py` and `protocols/chat_proto.py`) ```python def log_inbound(ctx: Context, msg_type: str, sender: str, details: str | None = None) -> None: suffix = f" {details}" if details else "" ctx.logger.info(f"[inbound] {msg_type} from {sender}{suffix}") def log_outbound(ctx: Context, msg_type: str, recipient: str, details: str | None = None) -> None: suffix = f" {details}" if details else "" ctx.logger.info(f"[outbound] {msg_type} -> {recipient}{suffix}") def log_state(ctx: Context, event: str, details: str | None = None) -> None: suffix = f" {details}" if details else "" ctx.logger.info(f"[state] {event}{suffix}") ``` Every `ctx.send(...)` in payment and chat code must be preceded by `log_outbound(...)`. Every `@payment_proto.on_message(...)` / `@chat_proto.on_message(...)` handler must start with `log_inbound(...)`. --- ## File 1 — `fet_payments/ledger.py` The only file that touches `cosmpy`. Pure functions, no `ctx`, no protocol types. ```python from __future__ import annotations import os from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError from typing import Any def extract_buyer_fet_wallet(metadata: Any) -> str | None: """Read the buyer's FET address from CommitPayment.metadata. Accepts either `buyer_fet_wallet` or `buyer_fet_address`. """ if not isinstance(metadata, dict): return None v = metadata.get("buyer_fet_wallet") or metadata.get("buyer_fet_address") return v if isinstance(v, str) and v else None def _env_int(name: str, default: int) -> int: try: return int(os.getenv(name, str(default))) except Exception: return default def verify_fet_payment_to_agent( *, transaction_id: str, expected_amount_fet: str, sender_fet_address: str, recipient_agent_wallet, # any object with .address() logger, use_mainnet: bool | None = None, ) -> bool: """Verify a direct FET transfer by inspecting the chain transaction. `expected_amount_fet` is in FET units (string, e.g. "0.001"). `recipient_agent_wallet` must expose `.address()` (e.g. `agent.wallet`). """ try: prefer_mainnet = (os.getenv("FET_USE_TESTNET", "true").strip().lower() != "true") networks: list[bool] = ( [use_mainnet] if use_mainnet is not None else [prefer_mainnet, not prefer_mainnet] ) for net_is_mainnet in networks: if _verify_fet_tx( transaction_id=transaction_id, expected_amount_fet=expected_amount_fet, sender_fet_address=sender_fet_address, recipient_agent_wallet=recipient_agent_wallet, logger=logger, use_mainnet=net_is_mainnet, ): return True return False except Exception as e: logger.error(f"FET payment verification failed: {e}") return False def _verify_fet_tx( *, transaction_id: str, expected_amount_fet: str, sender_fet_address: str, recipient_agent_wallet, logger, use_mainnet: bool, ) -> bool: expected_amount_micro = int(float(expected_amount_fet) * 10**18) denom = "afet" if use_mainnet else "atestfet" expected_recipient = str(recipient_agent_wallet.address()) logger.info( f"Verifying FET payment of {expected_amount_fet} FET ({expected_amount_micro} {denom}) " f"from {sender_fet_address} to {expected_recipient} " f"on {'mainnet' if use_mainnet else 'testnet'}" ) from cosmpy.aerial.client import LedgerClient, NetworkConfig network_config = ( NetworkConfig.fetchai_mainnet() if use_mainnet else NetworkConfig.fetchai_stable_testnet() ) ledger = LedgerClient(network_config) # `ledger.query_tx()` is blocking; enforce a timeout so we don't stall the event loop. query_timeout_s = float(_env_int("FET_LEDGER_QUERY_TIMEOUT_SECONDS", 20)) try: with ThreadPoolExecutor(max_workers=1) as ex: fut = ex.submit(ledger.query_tx, transaction_id) tx_resp = fut.result(timeout=query_timeout_s) except FuturesTimeoutError: logger.error( f"FET tx query timed out after {query_timeout_s:.0f}s on " f"{'mainnet' if use_mainnet else 'testnet'} (tx={transaction_id})" ) return False if tx_resp is None: logger.error(f"Transaction {transaction_id} not found on {'mainnet' if use_mainnet else 'testnet'}") return False if hasattr(tx_resp, "is_successful") and not tx_resp.is_successful(): logger.error(f"Transaction {transaction_id} was not successful") return False events = getattr(tx_resp, "events", None) if not isinstance(events, dict): logger.error(f"Transaction {transaction_id} has no usable events") return False valid_recipient = False valid_sender = False valid_amount = False transfer = events.get("transfer") if isinstance(transfer, dict): recipient = str(transfer.get("recipient") or "") sender = str(transfer.get("sender") or "") amount_str = str(transfer.get("amount") or "") if recipient == expected_recipient: valid_recipient = True if sender == sender_fet_address: valid_sender = True if amount_str.endswith(denom): try: if int(amount_str.replace(denom, "")) >= expected_amount_micro: valid_amount = True except ValueError: pass if not (valid_recipient and valid_amount): coin_received = events.get("coin_received") if isinstance(coin_received, dict): if str(coin_received.get("receiver") or "") == expected_recipient: valid_recipient = True amount_str = str(coin_received.get("amount") or "") if amount_str.endswith(denom): try: if int(amount_str.replace(denom, "")) >= expected_amount_micro: valid_amount = True except ValueError: pass if not valid_sender: coin_spent = events.get("coin_spent") if isinstance(coin_spent, dict) and str(coin_spent.get("spender") or "") == sender_fet_address: valid_sender = True if valid_recipient and valid_amount and valid_sender: logger.info(f"FET tx verified: {transaction_id}") return True logger.warning( "FET verification incomplete - " f"recipient={valid_recipient}, amount={valid_amount}, sender={valid_sender}" ) return False ``` --- ## File 2 — `protocols/payment_proto.py` Owns the `payment_protocol_spec` protocol, the request helper, and both handlers. Imports from `fet_payments.ledger` and exposes `set_agent_wallet`. ```python from __future__ import annotations import asyncio import os import time from uuid import uuid4 from uagents import Context, Protocol from uagents_core.contrib.protocols.payment import ( CancelPayment, CommitPayment, CompletePayment, Funds, RejectPayment, RequestPayment, payment_protocol_spec, ) from fet_payments.ledger import extract_buyer_fet_wallet, verify_fet_payment_to_agent payment_proto = Protocol(spec=payment_protocol_spec, role="seller") _agent_wallet = None # set from agent.py via set_agent_wallet(agent.wallet) def set_agent_wallet(wallet) -> None: global _agent_wallet _agent_wallet = wallet FET_AMOUNT_FET = (os.getenv("FET_AMOUNT_FET") or "0.001").strip() or "0.001" FET_FUNDS = Funds(currency="FET", amount=FET_AMOUNT_FET, payment_method="fet_direct") def _env_true(name: str, default: bool = True) -> bool: v = os.getenv(name) if v is None: return default return v.strip().lower() in {"1", "true", "yes", "y", "on"} def _use_testnet() -> bool: return os.getenv("FET_USE_TESTNET", "true").strip().lower() == "true" def log_inbound(ctx: Context, msg_type: str, sender: str, details: str | None = None) -> None: suffix = f" {details}" if details else "" ctx.logger.info(f"[inbound] {msg_type} from {sender}{suffix}") def log_outbound(ctx: Context, msg_type: str, recipient: str, details: str | None = None) -> None: suffix = f" {details}" if details else "" ctx.logger.info(f"[outbound] {msg_type} -> {recipient}{suffix}") def log_state(ctx: Context, event: str, details: str | None = None) -> None: suffix = f" {details}" if details else "" ctx.logger.info(f"[state] {event}{suffix}") def _latest_session_key(user_address: str) -> str: return f"payment_request:latest_session:{user_address}" def _fulfilled_key(user_address: str, session_id: str, tx_id: str) -> str: return f"payment:fulfilled:{user_address}:{session_id}:{tx_id}" def _resolve_chat_session_id(ctx: Context, user_address: str) -> str: """Best-effort session recovery for CommitPayment replays / session drift.""" try: rec = ctx.storage.get(_latest_session_key(user_address)) or {} if isinstance(rec, dict): sid = str(rec.get("session_id") or "").strip() if sid: return sid except Exception: pass return str(ctx.session) async def _chat(ctx: Context, user_address: str, text: str, *, reason: str | None = None) -> None: """Send a chat message without circular-importing chat_proto.""" from protocols.chat_proto import create_text_chat msg = create_text_chat(text) log_outbound(ctx, "ChatMessage", user_address, reason or "fet_payment_chat") await ctx.send(user_address, msg) async def request_payment_from_user( ctx: Context, user_address: str, description: str | None = None, text: str | None = None, ) -> None: """Send a FET RequestPayment to the user. `text` is the user's original prompt. We persist it under `chat_prompt::` so `on_commit` can fulfill the request after on-chain verification without a second user round-trip. """ if _agent_wallet is None: ctx.logger.error("[payment] agent wallet not set; call set_agent_wallet(agent.wallet)") return if not _env_true("ENABLE_FET_PAYMENTS", True): await _chat( ctx, user_address, "FET payments are currently disabled. Try again later.", reason="fet_payments_disabled", ) return session_id = str(ctx.session) pending_key = f"payment_request:pending:{user_address}:{session_id}" prompt_key = f"chat_prompt:{user_address}:{session_id}" now = time.time() # Always refresh the persisted prompt so the latest paid request wins. try: ctx.storage.set(prompt_key, text or "") ctx.storage.set(_latest_session_key(user_address), {"session_id": session_id, "ts": now}) except Exception as e: ctx.logger.warning(f"[payment] failed to persist chat prompt: {e}") pending_ttl_s = max(60, min(24 * 60 * 60, int(os.getenv("PAYMENT_REQUEST_PENDING_TTL_SECONDS", "1800")))) resend_min_s = max(10, min(3600, int(os.getenv("PAYMENT_REQUEST_RESEND_MIN_INTERVAL_SECONDS", "60")))) try: already_verified = bool(ctx.storage.get(f"{user_address}:{session_id}:verified_payment")) except Exception: already_verified = False if already_verified: log_state(ctx, "already_verified", f"user={user_address} session={session_id}") return try: pending = ctx.storage.get(pending_key) if ctx.storage.has(pending_key) else None except Exception: pending = None if isinstance(pending, dict): try: ts = float(pending.get("ts") or 0) last_sent = float(pending.get("last_sent") or 0) except Exception: ts, last_sent = 0.0, 0.0 if ts and (now - ts) < pending_ttl_s and (now - last_sent) < resend_min_s: log_state(ctx, "pending_request_suppressed", f"user={user_address} session={session_id}") return # don't spam the user use_testnet = _use_testnet() network_name = "stable-testnet" if use_testnet else "mainnet" metadata = { "fet_network": network_name, "mainnet": "false" if use_testnet else "true", "provider_agent_wallet": str(_agent_wallet.address()), } deadline_seconds = max(60, min(24 * 60 * 60, int(os.getenv("CHECKOUT_DEADLINE_SECONDS", "300")))) msg = RequestPayment( accepted_funds=[FET_FUNDS], recipient=str(_agent_wallet.address()), deadline_seconds=deadline_seconds, reference=str(uuid4()), description=description or "Please send the FET payment to continue.", metadata=metadata, ) log_outbound( ctx, "RequestPayment", user_address, f"amount={FET_FUNDS.amount} FET network={network_name} recipient={msg.recipient} ref={msg.reference}", ) await ctx.send(user_address, msg) try: ctx.storage.set(pending_key, { "ts": now, "last_sent": now, "reference": msg.reference, "deadline_seconds": deadline_seconds, "description": msg.description, "recipient": msg.recipient, "metadata": metadata, "accepted_funds": [{ "currency": FET_FUNDS.currency, "amount": FET_FUNDS.amount, "payment_method": FET_FUNDS.payment_method, }], }) log_state(ctx, "pending_request_saved", f"user={user_address} session={session_id} ref={msg.reference}") except Exception as e: ctx.logger.warning(f"[payment] failed to persist pending request: {e}") @payment_proto.on_message(CommitPayment) async def on_commit(ctx: Context, sender: str, msg: CommitPayment) -> None: tx_id = str(getattr(msg, "transaction_id", "") or "") method = str(getattr(msg.funds, "payment_method", "") or "") currency = str(getattr(msg.funds, "currency", "") or "") amount = str(getattr(msg.funds, "amount", "") or "") cancel_tx_id = tx_id or "missing_transaction_id" log_inbound( ctx, "CommitPayment", sender, f"tx_id={tx_id or 'MISSING'} method={method} currency={currency} amount={amount}", ) if _agent_wallet is None: ctx.logger.error("[payment] agent wallet not set; cannot verify") log_outbound(ctx, "CancelPayment", sender, "reason=server_wallet_missing") await ctx.send(sender, CancelPayment(transaction_id=cancel_tx_id, reason="Server wallet not configured")) return if method != "fet_direct" or currency != "FET": log_outbound(ctx, "CancelPayment", sender, f"reason=unsupported method={method} currency={currency}") await ctx.send( sender, CancelPayment(transaction_id=cancel_tx_id, reason=f"Unsupported payment method: {method}/{currency}"), ) return if not _env_true("ENABLE_FET_PAYMENTS", True): log_outbound(ctx, "CancelPayment", sender, "reason=fet_payments_disabled") await ctx.send(sender, CancelPayment(transaction_id=cancel_tx_id, reason="FET payments disabled")) return if not tx_id: log_outbound(ctx, "CancelPayment", sender, "reason=missing_tx_id") await ctx.send(sender, CancelPayment(transaction_id=cancel_tx_id, reason="Missing transaction_id")) return processed_key = f"payments:processed:{sender}:{tx_id}" resolved_session_id = _resolve_chat_session_id(ctx, sender) try: if ctx.storage.get(processed_key): fulfilled_key = _fulfilled_key(sender, resolved_session_id, tx_id) already_fulfilled = bool(ctx.storage.get(fulfilled_key)) log_state( ctx, "duplicate_commit", f"tx_id={tx_id} session={resolved_session_id} fulfilled={already_fulfilled}", ) log_outbound(ctx, "CompletePayment", sender, f"tx_id={tx_id} idempotent_replay=true") await ctx.send(sender, CompletePayment(transaction_id=tx_id)) if already_fulfilled: return # Verification already succeeded earlier, but fulfillment did not complete. try: prompt = str(ctx.storage.get(f"chat_prompt:{sender}:{resolved_session_id}") or "") except Exception: prompt = "" try: from protocols.chat_proto import run_paid_action await run_paid_action(ctx, sender, resolved_session_id, text=prompt) ctx.storage.set(fulfilled_key, True) log_state( ctx, "post_payment_fulfilled", f"user={sender} session={resolved_session_id} tx_id={tx_id} replay=true", ) except Exception as e: ctx.logger.exception(f"[payment] replay fulfillment failed: {e}") return except Exception: pass buyer_fet_wallet = extract_buyer_fet_wallet(msg.metadata) if not buyer_fet_wallet: log_outbound(ctx, "CancelPayment", sender, "reason=missing_buyer_fet_wallet") await ctx.send( sender, CancelPayment(transaction_id=cancel_tx_id, reason="Missing buyer_fet_wallet in metadata"), ) return use_testnet = _use_testnet() network_name = "stable-testnet" if use_testnet else "mainnet" ctx.logger.info( f"[payment] verifying FET tx tx_id={tx_id} network={network_name} " f"buyer={buyer_fet_wallet} recipient={_agent_wallet.address()} amount={amount}" ) try: verified = await asyncio.to_thread( verify_fet_payment_to_agent, transaction_id=tx_id, expected_amount_fet=str(msg.funds.amount), sender_fet_address=buyer_fet_wallet, recipient_agent_wallet=_agent_wallet, logger=ctx.logger, use_mainnet=not use_testnet, ) except Exception as e: ctx.logger.exception(f"[payment] FET verify error: {e}") verified = False ctx.logger.info(f"[payment] verify result tx_id={tx_id} verified={verified}") if not verified: log_outbound(ctx, "CancelPayment", sender, f"reason=verify_failed tx_id={tx_id}") await ctx.send( sender, CancelPayment(transaction_id=cancel_tx_id, reason="Payment verification failed"), ) return session_id = resolved_session_id fulfilled_key = _fulfilled_key(sender, session_id, tx_id) try: ctx.storage.set(processed_key, True) ctx.storage.set(f"{sender}:{session_id}:verified_payment", True) ctx.storage.remove(f"payment_request:pending:{sender}:{session_id}") log_state(ctx, "payment_verified", f"user={sender} session={session_id} tx_id={tx_id}") except Exception as e: ctx.logger.warning(f"[payment] failed to persist verified state: {e}") log_outbound(ctx, "CompletePayment", sender, f"tx_id={tx_id}") await ctx.send(sender, CompletePayment(transaction_id=tx_id)) try: prompt = str(ctx.storage.get(f"chat_prompt:{sender}:{session_id}") or "") except Exception: prompt = "" try: from protocols.chat_proto import run_paid_action await run_paid_action(ctx, sender, session_id, text=prompt) ctx.storage.set(fulfilled_key, True) log_state(ctx, "post_payment_fulfilled", f"user={sender} session={session_id} tx_id={tx_id}") except Exception as e: ctx.logger.exception(f"[payment] post-payment fulfillment failed: {e}") @payment_proto.on_message(RejectPayment) async def on_reject(ctx: Context, sender: str, msg: RejectPayment) -> None: reason = str(getattr(msg, "reason", "") or "no reason") log_inbound(ctx, "RejectPayment", sender, f"reason={reason[:120]}") session_id = _resolve_chat_session_id(ctx, sender) try: ctx.storage.remove(f"payment_request:pending:{sender}:{session_id}") log_state(ctx, "pending_request_cleared", f"user={sender} session={session_id}") except Exception: pass await _chat( ctx, sender, "You declined the payment. Reply with what you'd like to do and I'll send a new request.", reason="payment_reject_ack", ) ``` --- ## File 3 — `protocols/chat_proto.py` Trivial chat protocol that triggers payment. Replace `run_paid_action` with the real fulfillment for your agent. ```python from __future__ import annotations import asyncio import os from datetime import datetime, timezone from uuid import uuid4 from openai import OpenAI from uagents import Context, Protocol from uagents_core.contrib.protocols.chat import ( ChatAcknowledgement, ChatMessage, TextContent, chat_protocol_spec, ) from protocols.payment_proto import ( log_inbound, log_outbound, log_state, request_payment_from_user, ) chat_proto = Protocol(spec=chat_protocol_spec) ASI_ONE_BASE_URL = "https://api.asi1.ai/v1" ASI_ONE_MODEL = os.getenv("ASI_ONE_MODEL", "asi1").strip() or "asi1" ASI_ONE_SYSTEM_PROMPT = ( os.getenv("ASI_ONE_SYSTEM_PROMPT") or "You are a helpful, precise assistant. Answer the user's request directly." ) _asi_client: OpenAI | None = None def _get_asi_client() -> OpenAI: global _asi_client if _asi_client is None: _asi_client = OpenAI( api_key=os.getenv("ASI_ONE_API_KEY"), base_url=ASI_ONE_BASE_URL, ) return _asi_client def _call_asi_one(prompt: str, *, session_id: str) -> str: response = _get_asi_client().chat.completions.create( model=ASI_ONE_MODEL, messages=[ {"role": "system", "content": ASI_ONE_SYSTEM_PROMPT}, {"role": "user", "content": prompt}, ], temperature=0.2, max_tokens=1000, extra_headers={"x-session-id": session_id} if session_id else None, ) return response.choices[0].message.content or "(no response)" def create_text_chat(text: str) -> ChatMessage: return ChatMessage( timestamp=datetime.now(timezone.utc), msg_id=uuid4(), content=[TextContent(type="text", text=text)], ) def _user_text(msg: ChatMessage) -> str: parts = [c.text for c in msg.content if isinstance(c, TextContent)] return " ".join(parts).strip() def _is_paid_request(text: str) -> bool: """Replace this with your own gate (e.g. command, keyword, tool name).""" return bool(text) @chat_proto.on_message(ChatMessage) async def handle_chat(ctx: Context, sender: str, msg: ChatMessage) -> None: log_inbound(ctx, "ChatMessage", sender, f"msg_id={msg.msg_id}") ack = ChatAcknowledgement( timestamp=datetime.now(timezone.utc), acknowledged_msg_id=msg.msg_id, ) log_outbound(ctx, "ChatAcknowledgement", sender, f"acknowledged_msg_id={msg.msg_id}") await ctx.send(sender, ack) text = _user_text(msg) session_id = str(ctx.session) try: already_paid = bool(ctx.storage.get(f"{sender}:{session_id}:verified_payment")) except Exception: already_paid = False ctx.logger.info(f"[inbound] text from {sender}: {text[:120]!r} already_paid={already_paid}") if already_paid: await run_paid_action(ctx, sender, session_id, text=text) return if _is_paid_request(text): await request_payment_from_user(ctx, sender, description="Pay to run this request", text=text) return greet = create_text_chat("Send me a request to get started.") log_outbound(ctx, "ChatMessage", sender, "default_greeting") await ctx.send(sender, greet) @chat_proto.on_message(ChatAcknowledgement) async def handle_chat_ack(ctx: Context, sender: str, msg: ChatAcknowledgement) -> None: log_inbound(ctx, "ChatAcknowledgement", sender, f"acknowledged_msg_id={msg.acknowledged_msg_id}") async def run_paid_action( ctx: Context, user_address: str, session_id: str, text: str | None = None, ) -> None: """Run the paid action and reply on the chat protocol. Default implementation: route the prompt to ASI:One. Replace with image gen, video gen, a deterministic API call, etc., when the use case is not a chat completion. """ prompt = (text or "").strip() or "Say hello." try: reply_text = await asyncio.to_thread(_call_asi_one, prompt, session_id=session_id) except Exception as e: ctx.logger.exception(f"[payment] ASI:One call failed: {e}") reply_text = "Sorry — the assistant call failed. Please try again." reply = create_text_chat(reply_text) log_outbound(ctx, "ChatMessage", user_address, "paid_action_result") await ctx.send(user_address, reply) ``` --- ## Default LLM (ASI:One) `run_paid_action(...)` ships as a working ASI:One call out of the box. ASI:One is the **default** LLM for any agent scaffolded by this skill: it is OpenAI Chat Completions API-compatible, so the standard `openai` Python SDK works unchanged with `base_url` pointed at `https://api.asi1.ai/v1`. The wiring is already shown inline in File 3 — this section explains when to deviate. ### Decision: which LLM to use Walk the rules in order; pick the first match and stop. 1. **Does the paid action actually need an LLM?** If `run_paid_action` is image gen via a fixed endpoint, a deterministic API call, an echo, or anything non-LLM, replace the ASI:One body with the deterministic call and drop the `openai` import + `ASI_ONE_API_KEY`. 2. **Did the user explicitly name a different provider or model?** (e.g. `openai`, `gpt-4o`, `gpt-5`, `claude-3.5-sonnet`, `anthropic`, `gemini`, `groq`, `ollama`, `together`, `mistral`, etc.) — use exactly that and remove the ASI:One client. Treat `asi1`, `asi:one`, `asi-one`, `asione`, "ASI", or "ASI One" as an explicit confirmation of the default. 3. **Otherwise** — keep ASI:One. Do not require user opt-in. When extending an existing project that already wires up a different provider (`OPENAI_API_KEY`, `ANTHROPIC_API_KEY`, `langchain_*`, an `llm.py` wrapper, etc.), reuse that wiring instead of adding ASI:One on top — but for a **new** scaffold, ASI:One is the answer. Keep the LLM call confined to `protocols/chat_proto.py`. Do not import the OpenAI SDK from `payment_proto.py` or `fet_payments/ledger.py` — those files have a single concern each. ### Required env var (always) `.env.example` always includes `ASI_ONE_API_KEY`. Validate it at agent startup the same way `AGENT_SEED` is validated so the agent fails fast instead of silently 401-ing on the first paid request: ```python ASI_ONE_API_KEY = _required("ASI_ONE_API_KEY") ``` Do not add `OPENAI_API_KEY` (different provider) and never invent placeholder keys. ### Dependency (always) Always add the official `openai` Python SDK alongside `uagents` / `uagents-core`. Route the install through the active package skill (`uv add openai`, `poetry add openai`, `pip install openai`, or document-only for `no-package`). Do not pin a version unless the project already pins others. ### Optional knobs (use only when the use case needs them) ASI:One accepts every standard OpenAI parameter (`temperature`, `top_p`, `max_tokens`, `presence_penalty`, `frequency_penalty`, `stream`) plus two extras exposed via the OpenAI SDK's escape hatches: - **Web search** — `extra_body={"web_search": True}` to let the model browse, `False` to force it off. Omit otherwise. - **Agentic session persistence** — `extra_headers={"x-session-id": session_id}` to keep multi-turn agentic state across calls. Use a stable string per chat session (the `session_id` already passed into `run_paid_action` is a natural fit). ### Streaming ASI:One occasionally emits chunks with no `choices` or empty `delta.content`. If streaming, guard every chunk: ```python for chunk in response: if ( getattr(chunk, "choices", None) and chunk.choices and getattr(chunk.choices[0], "delta", None) and chunk.choices[0].delta.content ): partial = chunk.choices[0].delta.content ``` For a paid-action reply, batch streamed pieces into a single `ChatMessage` (or send incremental `ChatMessage`s with new `msg_id`s — the chat protocol does not support partial updates of an existing message). ### Response shape - `response.choices[0].message.content` — main reply text. Always read this for plain chat. - `response.usage` — token usage (`prompt_tokens`, `completion_tokens`, `total_tokens`). - `response.model` — model name echoed back. - ASI:One-specific fields (read defensively with `getattr`, may be absent): `executable_data` (Agentverse agent calls / tool manifests), `intermediate_steps` (multi-step reasoning trace), `thought` (model reasoning). ### Hard rules for ASI:One - Never hardcode the API key — always `os.getenv("ASI_ONE_API_KEY")`. - Never log the API key, full streamed text, or `intermediate_steps` / `thought` payloads — they may carry tool arguments or sensitive context. Log lengths only, consistent with the Directional Logging Contract above. - `base_url` is always exactly `https://api.asi1.ai/v1`. Do not call `https://api.openai.com` while routing to ASI:One. - Do not silently fall back to another provider if ASI:One is unreachable — surface the error and let the user decide. Silent fallback hides outages and bills the wrong account. - Keep the LLM call inside `protocols/chat_proto.py` (in `run_paid_action`). Do not put it in `payment_proto.py` or in `fet_payments/ledger.py`. --- ## File 4 — `agent.py` ```python from __future__ import annotations import os from dotenv import load_dotenv from uagents import Agent from uagents.setup import fund_agent_if_low from protocols.chat_proto import chat_proto from protocols.payment_proto import payment_proto, set_agent_wallet load_dotenv() def _required(name: str) -> str: v = os.getenv(name) if not v: raise RuntimeError(f"Missing required env var: {name}") return v # Validate critical env vars up front so the agent fails fast instead of # 401-ing on the first paid request. AGENT_SEED = _required("AGENT_SEED") ASI_ONE_API_KEY = _required("ASI_ONE_API_KEY") # Development default: testnet. This controls Almanac contract registration. # Switch to "mainnet" only when the user explicitly opts into a production deploy. # Note: this is independent of FET_USE_TESTNET, which controls the *payment* ledger # the agent verifies buyer transactions against. AGENT_NETWORK = (os.getenv("AGENT_NETWORK") or "testnet").strip().lower() agent = Agent( name=os.getenv("AGENT_NAME", "fet-paid-agent"), seed=AGENT_SEED, port=int(os.getenv("AGENT_PORT", "8000")), endpoint=os.getenv("AGENT_ENDPOINT") or None, mailbox=os.getenv("AGENT_MAILBOX", "true").strip().lower() == "true", network=AGENT_NETWORK, ) # REQUIRED on testnet: top up the agent wallet so Almanac registration succeeds. # Without this, uagents logs: # WARNING: [uagents.registration] I do not have enough funds to register on Almanac contract # WARNING: [uagents.registration] To enable contract registration, send funds to wallet address: fetch1... # fund_agent_if_low is a no-op once the wallet has enough testnet FET, so it is safe on every restart. # On mainnet the operator funds the wallet themselves; do not call it. if AGENT_NETWORK == "testnet": fund_agent_if_low(str(agent.wallet.address())) set_agent_wallet(agent.wallet) agent.include(chat_proto, publish_manifest=True) agent.include(payment_proto, publish_manifest=True) if __name__ == "__main__": agent.run() ``` --- ## File 5 — `.env.example` ```dotenv # Agent identity AGENT_NAME=fet-paid-agent AGENT_SEED=replace-with-a-strong-secret-seed AGENT_PORT=8000 AGENT_ENDPOINT= AGENT_MAILBOX=true # Almanac registration network. "testnet" auto-funds the wallet via fund_agent_if_low # and registers against the Fetch.ai testnet contract. Switch to "mainnet" only when # going to production (the user is then responsible for funding the wallet). # This is independent of FET_USE_TESTNET below, which controls the FET *payment* ledger. AGENT_NETWORK=testnet # Pricing (FET units, string) FET_AMOUNT_FET=0.001 # Network: "true" -> stable-testnet (atestfet), "false" -> mainnet (afet) FET_USE_TESTNET=true # Toggle FET payments (defaults to true) ENABLE_FET_PAYMENTS=true # Request UX CHECKOUT_DEADLINE_SECONDS=300 PAYMENT_REQUEST_PENDING_TTL_SECONDS=1800 PAYMENT_REQUEST_RESEND_MIN_INTERVAL_SECONDS=60 # Ledger query timeout (seconds) for cosmpy.query_tx FET_LEDGER_QUERY_TIMEOUT_SECONDS=20 # --- Default LLM (ASI:One, OpenAI-compatible) --- # ASI:One is the default LLM used by run_paid_action. Required at startup — # the agent fails fast if missing. See "Default LLM (ASI:One)" above for the # decision tree if you want to swap in a different provider. ASI_ONE_API_KEY= # Optional overrides: ASI_ONE_MODEL=asi1 ASI_ONE_SYSTEM_PROMPT=You are a helpful, precise assistant. Answer the user's request directly. ``` --- ## Funds + Metadata Shape (reference) `RequestPayment.accepted_funds[*]`: ```json { "currency": "FET", "amount": "0.001", "payment_method": "fet_direct" } ``` `RequestPayment.metadata`: ```json { "fet_network": "stable-testnet", "mainnet": "false", "provider_agent_wallet": "fetch1agentwallet..." } ``` `CommitPayment` (sent by the buyer/UI after broadcasting the on-chain transfer): ```json { "transaction_id": "", "funds": { "currency": "FET", "amount": "0.001", "payment_method": "fet_direct" }, "metadata": { "buyer_fet_wallet": "fetch1buyer..." } } ``` `buyer_fet_address` is also accepted as an alias for `buyer_fet_wallet`. --- ## Hard Rules - Resolve package manager via the **Package Skill Precedence** block before any setup/run instructions. - If a package skill is active, all setup/install/run commands must match that skill only. - Default `Agent(...)` to `network="testnet"` for development (driven by `AGENT_NETWORK=testnet` in `.env.example`); switch to `"mainnet"` only when the user explicitly opts into a production deploy. This is independent of `FET_USE_TESTNET`, which controls the FET *payment* ledger. - On `network="testnet"`, always call `fund_agent_if_low(str(agent.wallet.address()))` immediately after `Agent(...)` so the wallet has enough testnet FET for Almanac registration. This is what suppresses the `[uagents.registration] I do not have enough funds to register on Almanac contract` warning loop. On mainnet, do NOT call it — the operator funds the wallet themselves. - Never trust `CommitPayment` without on-ledger verification. Always call `verify_fet_payment_to_agent` (via `asyncio.to_thread`) before `CompletePayment`. - Reject FET commits without `buyer_fet_wallet` (or `buyer_fet_address`) in metadata. - Reject commits whose `funds.payment_method != "fet_direct"` or `funds.currency != "FET"`. - The seller agent's wallet must come from `agent.wallet`; deterministically derived from `AGENT_SEED`. Never hardcode addresses. - `expected_amount_fet` for verification must be the value from `msg.funds.amount`, not a server-side constant — this prevents under-payment via a tampered commit. - Use the chain `transaction_id` as the idempotency key (`payments:processed::`), and track a fulfillment marker (`payment:fulfilled:::`) so duplicate commits can safely resume fulfillment if the first attempt failed after verification. - Resolve commit session context from the latest pending paid session for the sender before falling back to `ctx.session` so prompt lookup and post-payment flow survive session drift/replay. - `FET_USE_TESTNET` selects the denom (`atestfet` vs `afet`); the verifier already double-checks both networks if needed, but the request metadata must report the chosen one so the wallet UI broadcasts on the right chain. - Wrap blocking ledger calls with `asyncio.to_thread` (or the `ThreadPoolExecutor` shown in File 1). Never `await` the verifier directly — it is synchronous. - Every handler starts with `log_inbound(...)`; every `ctx.send(...)` is preceded by `log_outbound(...)`; notable storage transitions use `log_state(...)`. ## Forbidden - Do not output mixed package-manager commands in one solution (for example `uv add ...` plus `pip install ...`). - Do not ship a "successful" path that skips verification when `cosmpy` is missing or times out. - Do not call `verify_fet_payment_to_agent(tx_hash=..., agent_wallet=..., expected_amount=...)` — those argument names are wrong and silently fail. The correct kwargs are `transaction_id`, `recipient_agent_wallet`, `expected_amount_fet`. - Do not put a top-level package called `fetchai/` or `cosmpy/` in your project — it shadows the SDK on import. - Do not log seeds, raw private keys, full signed transactions, or full `msg.metadata` dumps. - Do not call `ctx.send(...)` without a preceding `log_outbound(...)` line. - Do not import chat protocol from payment protocol at module top-level — use the local `_chat` helper to avoid circular imports. ## Idempotency Two independent layers protect against double-fulfillment: 1. **Request-side** — `pending_key = f"payment_request:pending:{user_address}:{session_id}"` plus `PAYMENT_REQUEST_PENDING_TTL_SECONDS` and `PAYMENT_REQUEST_RESEND_MIN_INTERVAL_SECONDS` prevent re-spamming `RequestPayment` while a previous one is still alive. 2. **Commit-side** — `processed_key = f"payments:processed:{sender}:{tx_id}"` is set the first time a `CommitPayment` is verified for that `tx_id`. Subsequent identical commits are idempotent replies (`CompletePayment`) and may safely resume `run_paid_action(...)` only when the fulfillment marker is missing. If you persist payment state in a database, also enforce a unique constraint on `(payment_reference, tx_hash)` and short-circuit on already-verified rows. ## Testing Hints - Use stable-testnet first (`FET_USE_TESTNET=true`). Fund the buyer wallet from the Fetch.ai testnet faucet, then send to `provider_agent_wallet` and call `CommitPayment` with the resulting tx hash. - Verify the happy path returns `CompletePayment` with the same `transaction_id`. - Negative cases that must produce `CancelPayment`: - `payment_method != "fet_direct"` or `currency != "FET"`. - Empty `transaction_id`. - Missing `buyer_fet_wallet` in metadata. - tx amount lower than `funds.amount`. - Recipient on the tx is not `provider_agent_wallet`. - Tx broadcast on the other network (testnet vs mainnet) — verifier will fail. - Re-send the same `CommitPayment` twice: - if fulfillment already finished: second call must log duplicate + emit `CompletePayment` idempotent replay only; - if fulfillment failed after verification: second call must resume `run_paid_action(...)` and log `post_payment_fulfilled ... replay=true`. - Send `RejectPayment` and confirm pending state is cleared (`[state] pending_request_cleared`) and the user gets a chat reply via `[outbound] ChatMessage -> payment_reject_ack`. ### Log Verification Checklist Run the agent locally (stable-testnet recommended) and confirm the CLI output shows the full trace: - `[inbound] ChatMessage from ` at entry, then `[outbound] ChatAcknowledgement -> `. - `[outbound] RequestPayment -> amount=... FET network=stable-testnet recipient=fetch1... ref=...`. - `[state] pending_request_saved user= session= ref=...` after persistence. - `[inbound] CommitPayment from tx_id=... method=fet_direct currency=FET amount=...` at entry. - `[payment] verifying FET tx tx_id=... network=stable-testnet buyer=fetch1... recipient=fetch1... amount=...`. - `[payment] verify result tx_id=... verified=true` on success. - `[state] payment_verified ...` then `[outbound] CompletePayment -> tx_id=...`. - Duplicate `CommitPayment` path emits `[state] duplicate_commit` + `idempotent_replay=true`; when replay resumes unfinished fulfillment it also logs `[state] post_payment_fulfilled ... replay=true`. - Failure path emits `[outbound] CancelPayment -> reason=...` with no raw error internals or secrets. - No seeds, raw private keys, full signed transactions, or unbounded `msg.metadata` dumps appear anywhere. ## Wrapping Existing Agents When adding FET payments to an agent that already speaks Chat Protocol: 1. Drop `fet_payments/ledger.py` in unchanged. 2. Add `protocols/payment_proto.py` and call `set_agent_wallet(agent.wallet)` in `agent.py` right after the `Agent(...)` construction. 3. `agent.include(payment_proto, publish_manifest=True)` next to `agent.include(chat_proto, ...)`. 4. In your existing chat handler, gate the paid branch with `ctx.storage.get(f"{sender}:{session_id}:verified_payment")` and call `request_payment_from_user(...)` when missing. 5. Replace the body of `run_paid_action` with the actual fulfillment. ## Further Reading (optional) - `uagents_core.contrib.protocols.payment`: `Funds`, `RequestPayment`, `CommitPayment`, `CompletePayment`, `CancelPayment`, `RejectPayment`, `payment_protocol_spec`. - `cosmpy.aerial.client.LedgerClient` and `NetworkConfig.fetchai_mainnet()` / `NetworkConfig.fetchai_stable_testnet()`. - The companion `chat-protocol` skill for `ChatMessage` / `ChatAcknowledgement` lifecycle. - The companion `stripe-payment-protocol` skill if you want to offer FET and cards side-by-side (add both Funds entries to `accepted_funds` and branch in `on_commit` by `payment_method`).