Version: Next

FET Image Agent Payment Protocol Example

This example showcases a paid image generation service where users pay with native FET tokens directly on the Fetch.ai blockchain. The agent verifies on-chain transactions before generating images using the HiDream AI model and delivers them through the chat protocol. This implementation demonstrates direct blockchain payment integration without third-party payment processors.

What it demonstrates

On-chain FET payment verification using Fetch.ai ledger queries.
Seller role implementation in the payment protocol workflow.
Integration with Replicate API for HiDream image generation.
Automatic image delivery as chat resources via temporary file hosting.
Robust error handling with retry mechanisms for external API calls.

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.py – Seller-side payment logic (request, verify FET payment, generate image)
client.py – HiDream image generation via Replicate API and tmpfiles.org upload
shared.py – Shared utilities for creating chat messages

Core Dependencies

The payment protocol requires these imports:

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

Additionally, for blockchain verification:

from cosmpy.aerial.client import LedgerClient, NetworkConfig

On-Chain Payment Verification

Unlike token-based payment systems, this example performs direct blockchain verification by querying transaction events on the Fetch.ai network:

payment.py
def check_fet_transaction_on_chain(
    tx_hash: str,
    required_amount: str,
    payer_address: str,
    agent_wallet,
    logger,
) -> bool:
    try:
        from cosmpy.aerial.client import LedgerClient, NetworkConfig
        testnet = os.getenv("FET_USE_TESTNET", "true").lower() == "true"
        network_config = (
            NetworkConfig.fetchai_stable_testnet() if testnet else NetworkConfig.fetchai_mainnet()
        )
        ledger = LedgerClient(network_config)

        # Convert FET amount to micro-denomination (atestfet/afet)
        amount_in_micro = int(float(required_amount) * 10**18)
        logger.info(
            f"Checking transaction {tx_hash} for {required_amount} FET "
            f"from {payer_address} to {agent_wallet.address()}"
        )

        # Query the transaction from the blockchain
        tx_result = ledger.query_tx(tx_hash)
        if not tx_result.is_successful():
            logger.error(f"Transaction {tx_hash} failed or not found")
            return False

        # Validate transfer event matches our requirements
        valid_recipient = False
        valid_amount = False
        valid_sender = False
        token_denom = "atestfet" if testnet else "afet"
        agent_address = str(agent_wallet.address())

        # Parse transaction events to find the transfer
        for event_name, event_data in tx_result.events.items():
            if event_name == "transfer":
                if event_data.get("recipient") == agent_address:
                    valid_recipient = True
                    if event_data.get("sender") == payer_address:
                        valid_sender = True
                    # Extract and validate amount
                    transfer_amount = event_data.get("amount", "")
                    if transfer_amount and transfer_amount.endswith(token_denom):
                        try:
                            micro_amount = int(transfer_amount.replace(token_denom, ""))
                            if micro_amount >= amount_in_micro:
                                valid_amount = True
                        except ValueError:
                            pass

        # All three conditions must be met
        if valid_recipient and valid_amount and valid_sender:
            logger.info(f"Transaction verified successfully: {tx_hash}")
            return True

        logger.warning(
            f"Verification incomplete - recipient: {valid_recipient}, "
            f"amount: {valid_amount}, sender: {valid_sender}"
        )
        return False
    except Exception as e:
        logger.error(f"FET payment verification failed: {e}")
        return False

Payment Request Handler

The seller agent initiates payment requests when users submit image prompts:

payment.py
import os
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 shared import create_text_chat

# Global wallet reference set by agent initialization
_agent_wallet = None

def initialize_payment_wallet(wallet):
    """Configure the agent's wallet for receiving payments"""
    global _agent_wallet
    _agent_wallet = wallet

# Create payment protocol with seller role
payment_proto = Protocol(spec=payment_protocol_spec, role="seller")

# Define accepted payment: 0.1 FET via direct on-chain transfer
PAYMENT_AMOUNT = Funds(currency="FET", amount="0.1", payment_method="fet_direct")

async def send_payment_request(ctx: Context, buyer_address: str, custom_message: str | None = None):
    """Create and send a payment request to the buyer."""
    
    payment_options = [PAYMENT_AMOUNT]
    request_metadata = {}
    
    # Configure network information for buyer
    network_name = "stable-testnet" if os.getenv("FET_USE_TESTNET", "true").lower() == "true" else "mainnet"
    if _agent_wallet:
        request_metadata["provider_agent_wallet"] = str(_agent_wallet.address())
    request_metadata["fet_network"] = network_name
    
    # Include custom message or default instructions
    if custom_message:
        request_metadata["content"] = custom_message
    else:
        request_metadata["content"] = (
            "Complete payment to unlock image generation. Your prompt will be processed after verification."
        )

    payment_request = RequestPayment(
        accepted_funds=payment_options,
        recipient=str(_agent_wallet.address()) if _agent_wallet else "unknown",
        deadline_seconds=300,  # 5 minute window
        reference=str(uuid4()),
        description=custom_message or "HiDream Image Generation: 0.1 FET per image",
        metadata=request_metadata
    )
    
    await ctx.send(buyer_address, payment_request)


@payment_proto.on_message(CommitPayment)
async def process_payment_commitment(ctx: Context, buyer: str, msg: CommitPayment):
    """Verify on-chain payment and trigger image generation if valid"""
    
    is_valid = False
    
    # Only process FET direct payments
    if msg.funds.payment_method == "fet_direct" and msg.funds.currency == "FET":
        try:
            # Extract buyer's FET wallet address from metadata
            buyer_wallet_addr = None
            if isinstance(msg.metadata, dict):
                buyer_wallet_addr = msg.metadata.get("buyer_fet_wallet") or msg.metadata.get("buyer_fet_address")
            
            if not buyer_wallet_addr:
                ctx.logger.error("Buyer FET wallet address missing in payment metadata")
                is_valid = False
            else:
                # Perform on-chain verification
                is_valid = check_fet_transaction_on_chain(
                    tx_hash=msg.transaction_id,
                    required_amount=PAYMENT_AMOUNT.amount,
                    payer_address=buyer_wallet_addr,
                    agent_wallet=_agent_wallet,
                    logger=ctx.logger,
                )
        except Exception as e:
            ctx.logger.error(f"Payment verification exception: {e}")
            is_valid = False
    
    if is_valid:
        ctx.logger.info(f"Payment confirmed from {buyer}, starting image generation")
        await ctx.send(buyer, CompletePayment(transaction_id=msg.transaction_id))
        await generate_image_after_payment(ctx, buyer)
    else:
        ctx.logger.warning(f"Payment verification failed for {buyer}")
        await ctx.send(
            buyer,
            CancelPayment(transaction_id=msg.transaction_id, reason="On-chain verification failed"),
        )

Image Generation Service

The agent uses Replicate's API to generate images with the HiDream model:

client.py
import os
import requests
from typing import Any, Optional
import aiohttp

# HiDream model identifier on Replicate
HIDREAM_MODEL = "prunaai/hidream-l1-fast:06898b39cb00e42d31666b0dc8b9904f326169768129d756184f65ecf1986c8f"
REPLICATE_API_TOKEN = os.getenv("REPLICATE_API_TOKEN")
TMPFILES_UPLOAD_URL = "https://tmpfiles.org/api/v1/upload"

def create_image_with_hidream(
    *,
    prompt: str,
    seed: int | None = None,
    speed_mode: str | None = None,
    output_format: str | None = None,
    output_quality: int | None = None,
) -> dict[str, Any] | None:
    """Generate image using HiDream via Replicate API with automatic retries."""
    api_key = os.getenv("REPLICATE_API_TOKEN")
    
    if not api_key:
        raise RuntimeError("REPLICATE_API_TOKEN environment variable required")

    model_inputs = {
        "seed": seed if seed is not None else 10,
        "prompt": prompt,
        "speed_mode": speed_mode if speed_mode is not None else "Extra Juiced 🚀 (even more speed)",
        "output_format": output_format if output_format is not None else "jpg",
        "output_quality": output_quality if output_quality is not None else 80,
    }

    # Call Replicate API with built-in retry mechanism
    api_response = _call_replicate_with_retries(HIDREAM_MODEL, model_inputs)
    
    # Extract image URL from response
    image_url = api_response.get("output")
    if image_url:
        # Handle both single URL and list responses
        url = image_url[0] if isinstance(image_url, list) else image_url
        http_response = requests.get(url, timeout=30)
        
        if http_response.status_code == 200:
            return {
                "image_data": http_response.content,
                "url": url,
                "content_type": http_response.headers.get('content-type', 'image/jpeg'),
                "status": "success"
            }
    
    return {"error": "Image generation failed", "status": "failed"}

async def store_image_temporarily(image_bytes: bytes, mime_type: str) -> Optional[str]:
    """Upload image to tmpfiles.org for temporary hosting (60 minute expiry)."""
    form_payload = aiohttp.FormData()
    form_payload.add_field('file', image_bytes, filename='generated_image.jpg', content_type=mime_type)
    
    async with aiohttp.ClientSession() as http_session:
        async with http_session.post(
            TMPFILES_UPLOAD_URL, 
            data=form_payload, 
            timeout=aiohttp.ClientTimeout(total=30)
        ) as response:
            if response.status == 200:
                json_result = await response.json()
                if json_result.get("status") == "success" and json_result.get("data"):
                    return json_result["data"].get("url", "")
    return None

Chat Message Processing

The chat protocol handles incoming messages and coordinates payment flow:

chat_proto.py
from datetime import datetime, timezone
from uagents import Context, Protocol
from uagents_core.contrib.protocols.chat import (
    ChatAcknowledgement,
    ChatMessage,
    TextContent,
    chat_protocol_spec,
)

from payment import (
    send_payment_request,
    generate_image_after_payment,
)
from shared import create_text_chat

chat_proto = Protocol(spec=chat_protocol_spec)

@chat_proto.on_message(ChatMessage)
async def process_chat_message(ctx: Context, user: str, msg: ChatMessage):
    # Acknowledge receipt of message
    await ctx.send(
        user,
        ChatAcknowledgement(
            timestamp=datetime.now(timezone.utc), acknowledged_msg_id=msg.msg_id
        ),
    )
    
    # Process text content
    for content_item in msg.content:
        if isinstance(content_item, TextContent):
            user_prompt = content_item.text.strip()
            current_session = str(ctx.session)
            
            # Storage keys for payment state tracking
            prompt_ready_key = f"{user}:{current_session}:awaiting_prompt"
            payment_confirmed_key = f"{user}:{current_session}:verified_payment"

            # Check if payment was already verified for this session
            payment_verified = (
                ctx.storage.has(prompt_ready_key) or ctx.storage.get(prompt_ready_key)
            ) and (
                ctx.storage.has(payment_confirmed_key) or ctx.storage.get(payment_confirmed_key)
            )

            if payment_verified:
                # Clear payment flags and generate image
                ctx.storage.remove(prompt_ready_key)
                ctx.storage.remove(payment_confirmed_key)
                ctx.storage.set(f"prompt:{user}:{current_session}", user_prompt)
                await generate_image_after_payment(ctx, user)
                return

            # New request: save prompt and request payment
            ctx.storage.set(f"prompt:{user}:{current_session}", user_prompt)
            await send_payment_request(
                ctx, 
                user, 
                custom_message="Payment required to generate your image. Please complete the transaction."
            )
            return

Helper Functions

Shared utilities for creating chat messages:

shared.py
from datetime import datetime, timezone
from uuid import uuid4
from uagents_core.contrib.protocols.chat import (
    AgentContent,
    ChatMessage,
    TextContent,
)

def create_text_chat(message: str, terminate_session: bool = False) -> ChatMessage:
    """Construct a chat message with optional session termination."""
    message_content: list[AgentContent] = [TextContent(type="text", text=message)]
    if terminate_session:
        from uagents_core.contrib.protocols.chat import EndSessionContent
        message_content.append(EndSessionContent(type="end-session"))
    return ChatMessage(
        timestamp=datetime.now(timezone.utc),
        msg_id=uuid4(),
        content=message_content,
    )

Agent Initialization

The main agent file sets up protocols and wallet:

agent.py
import dotenv
from uagents import Agent, Context

dotenv.load_dotenv()

from chat_proto import chat_proto
from payment import payment_proto, initialize_payment_wallet

# Create agent instance
agent = Agent()

# Register protocols with manifest publishing
agent.include(chat_proto, publish_manifest=True)
agent.include(payment_proto, publish_manifest=True)

# Configure wallet for receiving FET payments
initialize_payment_wallet(agent.wallet)

@agent.on_event("startup")
async def on_startup(ctx: Context):
    ctx.logger.info(f"HiDream Image Agent initialized: {agent.wallet.address()}")
    ctx.logger.info("=== FET-Paid Image Generation Service ===")
    ctx.logger.info("💳 Payment Method: 0.1 FET (on-chain)")
    ctx.logger.info("🤖 Model: HiDream via Replicate")
    ctx.logger.info("✅ Service ready to accept requests")

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

Configuration

Create a .env file with required credentials:

# Replicate API for image generation
REPLICATE_API_TOKEN=your_replicate_api_token_here

# Fetch.ai network selection
FET_USE_TESTNET=true  # Use 'false' for mainnet

# Optional: Agentverse connection
AGENTVERSE_URL=https://agentverse.ai

Installation & Execution

Set up the environment and run the agent:

# Create virtual environment
python3 -m venv .venv && source .venv/bin/activate

# Install dependencies
pip install uagents uagents-core cosmpy python-dotenv requests aiohttp

# Start the agent
python3 agent.py

Usage Flow

User sends message: Submit an image prompt via chat
Payment request: Agent responds with a payment request for 0.1 FET
User pays: Buyer completes on-chain FET transfer
Verification: Agent verifies transaction on Fetch.ai ledger
Generation: HiDream creates the image via Replicate API
Delivery: Image uploaded to tmpfiles.org and sent as chat resource

Implementation Details

Payment Verification Setup

Install blockchain library: pip install cosmpy
Network configuration: Set FET_USE_TESTNET in .env (true for testnet, false for mainnet)
Payment definition: In payment.py, define PAYMENT_AMOUNT = Funds(currency="FET", amount="0.1", payment_method="fet_direct")
Buyer metadata: On CommitPayment, buyer must provide buyer_fet_wallet in msg.metadata
On-chain check: Agent uses LedgerClient.query_tx() to verify transaction events match expected transfer

Key Implementation Features

Blockchain Integration: Direct on-chain verification without payment intermediaries
Resilient API Calls: Automatic retries with exponential backoff for Replicate and tmpfiles.org
State Management: Session-based storage tracks payment status and prompts
Error Recovery: Failed generations allow one retry without additional payment
Resource Protocol: Images delivered as ResourceContent with temporary hosting URLs

Getting FET Testnet Tokens

To test this agent on the Fetch.ai testnet, you'll need testnet FET tokens in your wallet. Here's how to get them:

Visit the Testnet Faucet: Go to the official Fetch.ai Testnet Faucet to request testnet tokens
Wait for Confirmation: Testnet tokens are typically distributed automatically after a short delay

For detailed instructions and a step-by-step visual guide, check out this blog post or watch the video below:

note

Get the full example with implementation here.

What it demonstrates​

Project Structure​

Core Dependencies​

On-Chain Payment Verification​

Payment Request Handler​

Image Generation Service​

Chat Message Processing​

Helper Functions​

Agent Initialization​

Configuration​

Installation & Execution​

Usage Flow​

Implementation Details​

Payment Verification Setup​

Key Implementation Features​

Getting FET Testnet Tokens​