Solana Wallet Balance Agent

This guide demonstrates how to create a Solana Wallet Balance Agent that can check wallet balances using the Solana RPC API. The agent is compatible with ASI1 LLM and can process natural language queries about Solana wallet balances.

Overview

The Solana Wallet Balance Agent allows users to query wallet balances using natural language. It uses the Solana RPC API to fetch real-time balance information and provides formatted responses with both SOL and lamports values.

Message Flow

The communication flow between ASI1 LLM, the Solana Wallet Agent, and OpenAI Agent follows this sequence:

1. Query Initiation (1.1)

   • ASI1 LLM sends a natural language query (e.g., "What's the balance of wallet address AtTjQKXo1CYTa2MuxPARtr382ZyhPU5YX4wMMpvaa1oy") as a ChatMessage to the Solana Wallet Agent.

2. Parameter Extraction (2, 3)

   • The Solana Wallet Agent forwards the query to OpenAI Agent for parameter extraction
   • OpenAI Agent processes the natural language and extracts the wallet address
   • The address is returned in a Pydantic Model format as StructuredOutputResponse

3. Balance Query (4, 5)

   • The Solana Wallet Agent calls the get_balance_from_address function with the extracted address
   • The function queries the Solana RPC API and returns the balance information

4. Agent Response (6.1)

   • The Solana Wallet Agent sends the formatted response back as a ChatMessage to ASI1 LLM

5. Message Acknowledgements (1.2, 6.2)

   • Each message is acknowledged using ChatAcknowledgement

Here's a visual representation of the flow:

Implementation

In this example we will create an agent and its associated files on Agentverse that communicate using the chat protocol with the ASI1 LLM. Refer to the Hosted Agents section to understand the detailed steps for agent creation on Agentverse.

Create a new agent named "SolanaWalletAgent" on Agentverse and create the following files:

agent.py            # Main agent file with integrated chat protocol and message handlers for ChatMessage and ChatAcknowledgement
solana_service.py   # Solana RPC API integration

To create a new file on Agentverse:

1. Click on the New File icon

2. Assign a name to the File

3. Directory Structure

1. Solana Service Implementation

The solana_service.py file handles the interaction with the Solana RPC API:

solana_service.py

import os
import logging
import requests
import json
from uagents import Model, Field

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Solana RPC endpoint
SOLANA_RPC_URL = "https://api.mainnet-beta.solana.com"

class SolanaRequest(Model):
    address: str = Field(
        description="Solana wallet address to check",
    )

class SolanaResponse(Model):
    balance: str = Field(
        description="Formatted Solana wallet balance",
    )

async def get_balance_from_address(address: str) -> str:
    """
    Get the balance for a Solana address using the Solana RPC API
    
    Args:
        address: Solana wallet address
        
    Returns:
        Formatted balance string
    """
    try:
        logger.info(f"Getting balance for address: {address}")
        
        # Prepare the request payload
        payload = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "getBalance",
            "params": [address]
        }
        
        # Set headers
        headers = {
            "Content-Type": "application/json"
        }
        
        # Make the API request
        response = requests.post(SOLANA_RPC_URL, headers=headers, json=payload)
        response.raise_for_status()
        
        # Parse the response
        result = response.json()
        
        if "error" in result:
            error_msg = f"Error: {result['error']['message']}"
            logger.error(error_msg)
            return error_msg
            
        if "result" in result and "value" in result["result"]:
            # Convert lamports to SOL (1 SOL = 1,000,000,000 lamports)
            lamports = result["result"]["value"]
            sol_balance = lamports / 1_000_000_000
            
            # Format the result
            result_str = f"{sol_balance:.9f} SOL ({lamports} lamports)"
            logger.info(f"Balance for {address}: {result_str}")
            return result_str
        else:
            error_msg = "No balance information found"
            logger.error(error_msg)
            return error_msg
            
    except requests.exceptions.RequestException as e:
        error_msg = f"Request error: {str(e)}"
        logger.error(error_msg)
        return error_msg
    except json.JSONDecodeError as e:
        error_msg = f"JSON decode error: {str(e)}"
        logger.error(error_msg)
        return error_msg
    except Exception as e:
        error_msg = f"Unexpected error: {str(e)}"
        logger.error(error_msg)
        return error_msg

2. Solana Wallet Agent Setup

The agent.py file is the core of your application with integrated chat protocol functionality and contains message handlers for ChatMessage and ChatAcknowledgement protocols. Think of it as the main control center that:

• Sets up your agent and handles chat protocol messages with dedicated handlers for processing wallet balance queries
• Manages rate limiting and monitors the agent's health for optimal Solana API performance

agent.py

import os
from enum import Enum
from datetime import datetime, timezone
from uuid import uuid4
from typing import Any

from uagents import Agent, Context, Model, Protocol
from uagents.experimental.quota import QuotaProtocol, RateLimit
from uagents_core.models import ErrorMessage

# Import chat protocol components
from uagents_core.contrib.protocols.chat import (
    chat_protocol_spec,
    ChatMessage,
    ChatAcknowledgement,
    TextContent,
    EndSessionContent,
    StartSessionContent,
)

from solana_service import get_balance_from_address, SolanaRequest, SolanaResponse

agent = Agent()

# AI Agent for structured output (choose one)
AI_AGENT_ADDRESS = 'agent1q0h70caed8ax769shpemapzkyk65uscw4xwk6dc4t3emvp5jdcvqs9xs32y'  # OpenAI Agent
# AI_AGENT_ADDRESS = 'agent1qvk7q2av3e2y5gf5s90nfzkc8a48q3wdqeevwrtgqfdl0k78rspd6f2l4dx'  # Claude Agent

if not AI_AGENT_ADDRESS:
    raise ValueError("AI_AGENT_ADDRESS not set")

# Create the chat protocol
chat_proto = Protocol(spec=chat_protocol_spec)

# Create structured output protocol
struct_output_client_proto = Protocol(
    name="StructuredOutputClientProtocol", version="0.1.0"
)

# Structured output models
class StructuredOutputPrompt(Model):
    prompt: str
    output_schema: dict[str, Any]

class StructuredOutputResponse(Model):
    output: dict[str, Any]

def create_text_chat(text: str, end_session: bool = True) -> ChatMessage:
    content = [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 protocol message handler
@chat_proto.on_message(ChatMessage)
async def handle_message(ctx: Context, sender: str, msg: ChatMessage):
    ctx.logger.info(f"Got a message from {sender}: {msg.content}")
    ctx.storage.set(str(ctx.session), sender)
    
    # Send acknowledgement
    await ctx.send(
        sender,
        ChatAcknowledgement(
            acknowledged_msg_id=msg.msg_id, 
            timestamp=datetime.now(timezone.utc)
        ),
    )

    # Process message content
    for content in msg.content:
        if isinstance(content, StartSessionContent):
            ctx.logger.info(f"Got a start session message from {sender}")
            continue
        elif isinstance(content, TextContent):
            ctx.logger.info(f"Got a message from {sender}: {content.text}")
            ctx.storage.set(str(ctx.session), sender)
            
            # Send to AI agent for structured output extraction
            await ctx.send(
                AI_AGENT_ADDRESS,
                StructuredOutputPrompt(
                    prompt=content.text, 
                    output_schema=SolanaRequest.schema()
                ),
            )
        else:
            ctx.logger.info(f"Got unexpected content from {sender}")

# Handle structured output response from AI agent
@struct_output_client_proto.on_message(StructuredOutputResponse)
async def handle_structured_output_response(
    ctx: Context, sender: str, msg: StructuredOutputResponse
):
    session_sender = ctx.storage.get(str(ctx.session))
    if session_sender is None:
        ctx.logger.error(
            "Discarding message because no session sender found in storage"
        )
        return

    if "<UNKNOWN>" in str(msg.output):
        await ctx.send(
            session_sender,
            create_text_chat(
                "Sorry, I couldn't process your request. Please include a valid Solana wallet address."
            ),
        )
        return

    try:
        # Parse the structured output to get the address
        wallet_request = SolanaRequest.parse_obj(msg.output)
        address = wallet_request.address
        
        if not address:
            await ctx.send(
                session_sender,
                create_text_chat(
                    "Sorry, I couldn't find a valid Solana wallet address in your query."
                ),
            )
            return
        
        # Get the balance for this address
        balance = await get_balance_from_address(address)
        
        # Create a nicely formatted response
        response_text = f"Wallet Balance for `{address}`:\n{balance}\n\n[View on Solana Explorer](https://explorer.solana.com/address/{address})"
        
        # Send the response back to the user
        await ctx.send(session_sender, create_text_chat(response_text))
        
    except Exception as err:
        ctx.logger.error(f"Error processing structured output: {err}")
        await ctx.send(
            session_sender,
            create_text_chat(
                "Sorry, I couldn't check the wallet balance. Please try again later."
            ),
        )

# Chat protocol acknowledgement handler
@chat_proto.on_message(ChatAcknowledgement)
async def handle_ack(ctx: Context, sender: str, msg: ChatAcknowledgement):
    ctx.logger.info(
        f"Got an acknowledgement from {sender} for {msg.acknowledged_msg_id}"
    )

# Optional: Rate limiting protocol for direct requests
proto = QuotaProtocol(
    storage_reference=agent.storage,
    name="Solana-Wallet-Protocol",
    version="0.1.0",
    default_rate_limit=RateLimit(window_size_minutes=60, max_requests=30),
)

# Optional: Direct request handler for structured requests
@proto.on_message(SolanaRequest, replies={SolanaResponse, ErrorMessage})
async def handle_request(ctx: Context, sender: str, msg: SolanaRequest):
    ctx.logger.info(f"Received wallet balance request for address: {msg.address}")
    try:
        balance = await get_balance_from_address(msg.address)
        ctx.logger.info(f"Successfully fetched wallet balance for {msg.address}")
        await ctx.send(sender, SolanaResponse(balance=balance))
    except Exception as err:
        ctx.logger.error(err)
        await ctx.send(sender, ErrorMessage(error=str(err)))

# Optional: Health check related code
def agent_is_healthy() -> bool:
    """
    Implement the actual health check logic here.
    For example, check if the agent can connect to the Solana RPC API.
    """
    try:
        import asyncio
        asyncio.run(get_balance_from_address("AtTjQKXo1CYTa2MuxPARtr382ZyhPU5YX4wMMpvaa1oy"))
        return True
    except Exception:
        return False

class HealthCheck(Model):
    pass

class HealthStatus(str, Enum):
    HEALTHY = "healthy"
    UNHEALTHY = "unhealthy"

class AgentHealth(Model):
    agent_name: str
    status: HealthStatus

health_protocol = QuotaProtocol(
    storage_reference=agent.storage, name="HealthProtocol", version="0.1.0"
)

@health_protocol.on_message(HealthCheck, replies={AgentHealth})
async def handle_health_check(ctx: Context, sender: str, msg: HealthCheck):
    status = HealthStatus.UNHEALTHY
    try:
        if agent_is_healthy():
            status = HealthStatus.HEALTHY
    except Exception as err:
        ctx.logger.error(err)
    finally:
        await ctx.send(sender, AgentHealth(agent_name="solana_wallet_agent", status=status))

# Register protocols
agent.include(chat_proto, publish_manifest=True)
agent.include(struct_output_client_proto, publish_manifest=True)
agent.include(proto, publish_manifest=True)
agent.include(health_protocol, publish_manifest=True)

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

Adding a README to your Agent

1. Go to the Overview section in the Editor.

2. Click on Edit and add a good description for your Agent so that it can be easily searchable by the ASI1 LLM. Please refer the Importance of Good Readme section for more details.

3. Make sure the Agent has the right AgentChatProtocol.

Query your Agent from ASI1 LLM

1. Login to the ASI1 LLM, either using your Google Account or the ASI1 Wallet and Start a New Chat.

2. Toggle the "Agents" switch to enable ASI1 to connect with Agents on Agentverse.

3. Type in a query to ask your Agent for instance 'What's the balance of wallet address 6wFKPxNToSnggrZr4P4s1r4zRxuJX2nSA7iTdQDPpgHc'.

Note: The ASI1 LLM may not always select your agent for answering the query as it is designed to pick the best agent for a task based on a number of parameters.