Image Agent Payment Protocol Example

This example demonstrates a seller agent that requests a small payment (USDC via Skyfire) before generating an image and sending it back as a resource using the chat protocol. It is adapted from your image-agent-payment-protocol project.

What it shows

• Minimal payment workflow using AgentPaymentProtocol with roles.
• Skyfire token verification and charge flow.
• Image generation via Pollinations.
• Returning the image through the chat protocol as a resource.

Project Structure

Key files used in this example:

• agent.py – Agent setup; includes chat and payment protocols
• chat_proto.py – Chat protocol and message handlers
• payment_proto.py – Seller-side payment logic (request, verify, charge)
• skyfire.py – JWT verification and charge against Skyfire API

Payment Protocol (imports)

from uagents_core.contrib.protocols.payment import (
    Funds,
    RequestPayment,
    RejectPayment,
    CommitPayment,
    CancelPayment,
    CompletePayment,
    payment_protocol_spec,
)

Skyfire helpers (pseudocode)

Click here to get the full file.

skyfire.py (pseudocode)

# Env inputs: SKYFIRE_API_KEY, SKYFIRE_SERVICE_ID, SELLER_ACCOUNT_ID or JWT_AUDIENCE,
#             JWKS_URL, JWT_ISSUER, SKYFIRE_TOKENS_API_URL

async def verify_token_claims(token, logger) -> bool:
    # 1) Fetch JWKS from JWKS_URL
    # 2) Decode JWT with issuer=JWT_ISSUER and audience=JWT_AUDIENCE or SELLER_ACCOUNT_ID
    # 3) Ensure claim ssi == SKYFIRE_SERVICE_ID
    # Return True on success, else False
    pass

async def charge_token(token, amount_usdc, logger) -> bool:
    # POST SKYFIRE_TOKENS_API_URL
    #   headers: { "skyfire-api-key": SKYFIRE_API_KEY, "skyfire-api-version": "2" }
    #   body:    { "token": token, "chargeAmount": amount_usdc }
    # Return True on 2xx, else False
    pass

async def verify_and_charge(token, amount_usdc, logger) -> bool:
    # Ensure required env vars exist
    # If verify_token_claims(token): return charge_token(token, amount_usdc)
    # Else return False
    pass

def get_skyfire_service_id() -> str | None:
    # Return SKYFIRE_SERVICE_ID from env
    pass

Payment Logic (seller)

payment_proto.py

import os
from uuid import uuid4
from uagents import Context, Protocol

from uagents_core.contrib.protocols.payment import (
    Funds,
    RequestPayment,
    RejectPayment,
    CommitPayment,
    CancelPayment,
    CompletePayment,
    payment_protocol_spec,
)
from skyfire import verify_and_charge, get_skyfire_service_id
from chat_proto import create_text_chat
from urllib.parse import quote


_agent_wallet = None


def set_agent_wallet(wallet):
    global _agent_wallet
    _agent_wallet = wallet


payment_proto = Protocol(spec=payment_protocol_spec, role="seller")

USDC_FUNDS = Funds(currency="USDC", amount="0.001", payment_method="skyfire")


async def request_payment_from_user(ctx: Context, user_address: str):
    accepted_funds = [USDC_FUNDS]
    skyfire_service_id = get_skyfire_service_id()

    metadata = {}
    if skyfire_service_id:
        metadata["skyfire_service_id"] = skyfire_service_id
    if _agent_wallet:
        metadata["provider_agent_wallet"] = str(_agent_wallet.address())

    payment_request = RequestPayment(
        accepted_funds=accepted_funds,
        recipient=ctx.agent.address,
        deadline_seconds=300,
        reference=str(uuid4()),
        description="ASI1 Image Gen: after payment, send your image prompt (one image per payment)",
        metadata=metadata,
    )

    await ctx.send(user_address, payment_request)


@payment_proto.on_message(CommitPayment)
async def handle_commit_payment(ctx: Context, sender: str, msg: CommitPayment):
    payment_verified = False
    if msg.funds.payment_method == "skyfire" and msg.funds.currency == "USDC":
        payment_verified = await verify_and_charge(msg.transaction_id, "0.001", ctx.logger)

    if payment_verified:
        session_id = str(ctx.session)
        ctx.storage.set(f"{sender}:{session_id}:awaiting_prompt", True)
        ctx.storage.set(f"{sender}:{session_id}:verified_payment", True)
        await ctx.send(sender, CompletePayment(transaction_id=msg.transaction_id))
        await ctx.send(sender, create_text_chat("Payment verified. Please send your image prompt."))
    else:
        await ctx.send(sender, RejectPayment(reason="Payment verification failed"))

Chat Protocol integration

chat_proto.py

from datetime import datetime, timezone
from uuid import uuid4

from uagents import Context, Protocol
from uagents_core.contrib.protocols.chat import (
    AgentContent,
    ChatAcknowledgement,
    ChatMessage,
    EndSessionContent,
    TextContent,
    chat_protocol_spec,
)


def create_text_chat(text: str, end_session: bool = False) -> ChatMessage:
    content: list[AgentContent] = [TextContent(type="text", text=text)]
    if end_session:
        content.append(EndSessionContent(type="end-session"))
    return ChatMessage(
        timestamp=datetime.now(timezone.utc),
        msg_id=uuid4(),
        content=content,
    )


chat_proto = Protocol(spec=chat_protocol_spec)


@chat_proto.on_message(ChatMessage)
async def handle_message(ctx: Context, sender: str, msg: ChatMessage):
    from payment_proto import request_payment_from_user

    for item in msg.content:
        if isinstance(item, TextContent):
            await ctx.send(sender, create_text_chat("Please complete a small payment first."))
            await request_payment_from_user(ctx, sender)

Agent setup

agent.py

import os
import dotenv
from uagents import Agent, Context

dotenv.load_dotenv()

from chat_proto import chat_proto
from payment_proto import payment_proto, set_agent_wallet


agent = Agent(
    name="ASI1ImageAgent",
    port=8021,
    mailbox=True,
    agentverse=os.getenv("AGENTVERSE_URL", "https://agentverse.ai"),
)

set_agent_wallet(agent.wallet)

@agent.on_event("startup")
async def startup(ctx: Context):
    ctx.logger.info(f"ASI1 Image Agent started: {agent.wallet.address()}")

agent.include(chat_proto, publish_manifest=True)
agent.include(payment_proto, publish_manifest=True)

if __name__ == "__main__":
    agent.run()

Environment

Prepare a .env with your Skyfire credentials and Agentverse settings:

AGENTVERSE_URL=https://agentverse.ai
SKYFIRE_API_KEY=
SKYFIRE_SERVICE_ID=
SELLER_ACCOUNT_ID=
JWKS_URL=https://app.skyfire.xyz/.well-known/jwks.json
JWT_ISSUER=https://app.skyfire.xyz/
SKYFIRE_TOKENS_API_URL=https://api.skyfire.xyz/api/v1/tokens/charge

Run locally

python3 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
python3 agent.py

Send a short chat message to the agent; it will respond with a payment request. After committing payment, it will prompt for the image description and then return the generated image as a resource.

Enabling Skyfire payments

1. Place skyfire.py next to your agent.py and payment_proto.py.
2. Set environment variables in .env:
   • SKYFIRE_API_KEY, SKYFIRE_SERVICE_ID, and either SELLER_ACCOUNT_ID or JWT_AUDIENCE
   • Optional overrides: JWKS_URL, JWT_ISSUER, SKYFIRE_TOKENS_API_URL
3. In payment_proto.py, keep USDC_FUNDS = Funds(currency="USDC", amount="0.001", payment_method="skyfire") and surface skyfire_service_id in RequestPayment.metadata.
4. On CommitPayment, call verify_and_charge(msg.transaction_id, "0.001", ctx.logger) and only proceed to CompletePayment on success.

note

Get the full example with implementation here.