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
solana_service.py   # Solana RPC API integration
chat_proto.py       # Chat protocol implementation

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. Chat Protocol Integration

The chat_proto.py file is essential for enabling natural language communication between your agent and ASI1 LLM.

LLM Integration

To extract the important parameters from a user query passed by the LLM, you can use any of the following LLMs that support structured output:

• OpenAI Agent: agent1q0h70caed8ax769shpemapzkyk65uscw4xwk6dc4t3emvp5jdcvqs9xs32y
• Claude.ai Agent: agent1qvk7q2av3e2y5gf5s90nfzkc8a48q3wdqeevwrtgqfdl0k78rspd6f2l4dx

Note: To ensure fair usage, each agent is limited to 6 requests per hour. Please implement appropriate rate limiting in your application.

Message Flow

1. When a user sends a query (e.g., "What's the balance of wallet address AtTjQKXo1CYTa2MuxPARtr382ZyhPU5YX4wMMpvaa1oy"):

   • The ChatMessage handler receives the message
   • Acknowledges receipt to the sender
   • Forwards the query to the chosen LLM for processing

2. The LLM processes the query and returns a StructuredOutputResponse:

   • Extracts relevant parameters (e.g., address="AtTjQKXo1CYTa2MuxPARtr382ZyhPU5YX4wMMpvaa1oy")
   • Returns data in the format specified by your agent's schema

3. Your agent processes the structured data:

   • Calls appropriate functions (e.g., get_balance_from_address)
   • Formats the response and sends it back.

Customizing the Handlers

When implementing a new agent for a different use case, you'll need to:

• Update the schema in StructuredOutputPrompt to match your agent's data model
• Modify the response handling in handle_structured_output_response function for your use case.

To enable natural language communication with your agent add the following chat protocol in the chat_proto.py file created on Agentverse:

chat_proto.py

from datetime import datetime
from uuid import uuid4
from typing import Any

from uagents import Context, Model, Protocol

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

from solana_service import get_balance_from_address, SolanaRequest

# AI Agent Address for structured output processing
AI_AGENT_ADDRESS = 'agent1q0h70caed8ax769shpemapzkyk65uscw4xwk6dc4t3emvp5jdcvqs9xs32y'

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

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.utcnow(),
        msg_id=uuid4(),
        content=content,
    )

chat_proto = Protocol(spec=chat_protocol_spec)
struct_output_client_proto = Protocol(
    name="StructuredOutputClientProtocol", version="0.1.0"
)

class StructuredOutputPrompt(Model):
    prompt: str
    output_schema: dict[str, Any]

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

@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}")
    ctx.storage.set(str(ctx.session), sender)
    await ctx.send(
        sender,
        ChatAcknowledgement(timestamp=datetime.utcnow(), acknowledged_msg_id=msg.msg_id),
    )

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

@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}"
    )

@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(err)
        await ctx.send(
            session_sender,
            create_text_chat(
                "Sorry, I couldn't check the wallet balance. Please try again later."
            ),
        )
        return

3. Solana Wallet Agent Setup

The agent.py file is the core of your application. Think of it as the main control center that:

• Sets up your agent
• Handles incoming requests
• Manages rate limiting
• Monitors the agent's health

Let's break down each component:

Basic Setup

from uagents import Agent, Context, Model
from uagents.experimental.quota import QuotaProtocol, RateLimit

agent = Agent()  # Create a new agent instance

Import the necessary packages and initialise your agent.

Rate Limiting Setup

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),
)

To ensure fair usage and prevent API abuse, we recommend implementing the QuotaProtocol. While optional, this protocol helps manage service usage by limiting requests. In this example, we've set a limit of 30 requests per hour per user, but you can adjust this threshold based on your agent's functionality. This helps maintain service reliability and protects your agent from excessive requests.

Request Handler

@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)))

This message handler allows your agent to receive direct requests from other agents in a structured format. While we're using it with ASI1 LLM in this example, it's versatile enough to work with agents that don't have the chat protocol enabled. This makes it particularly useful in multi-agent systems, where other agents can directly request information directly.

Health Monitoring

### 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))

The HealthProtocol, while optional, provides a periodic health check system to monitor your agent's performance and reliability. It tests the agent's ability to fetch information about a wallet balance and verifies that all components are working correctly. This monitoring helps quickly identify and address any issues that might affect your agent's service, ensuring reliable operation for your users.

Protocol Registration

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

This registers all the necessary protocols with your agent:

• Main protocol for handling wallet balance requests
• Health monitoring protocol
• Chat protocol for natural language communication
• Structured output protocol for formatted responses

Here's the complete implementation:

agent.py

import os
from enum import Enum

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

from chat_proto import chat_proto, struct_output_client_proto
from solana_service import get_balance_from_address, SolanaRequest, SolanaResponse

agent = Agent()

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),
)

@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)))

agent.include(proto, publish_manifest=True)

### 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))

agent.include(health_protocol, publish_manifest=True)
agent.include(chat_proto, publish_manifest=True)
agent.include(struct_output_client_proto, 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.