Solana Wallet Balance Agent

This guide shows how to build a read-only Solana wallet balance agent. Users ask in natural language (for example, “What is the SOL balance of this wallet?”). The agent extracts an address, queries Solana JSON-RPC, and replies with SOL plus lamports and an Explorer link.

It speaks the Agent Chat Protocol so ASI:One can route those questions to your agent.

This page is balance lookups only. It does not sign transactions or transfer SOL. For wallet signing and on-chain transfers, see Solana agent integration (devnet patterns).

The published slug and frontmatter id are solana-wallet-agent (/docs/examples/chat-protocol/solana-wallet-agent). The source filename is solana-balance-check.mdx; Docusaurus uses the id, not the filename, for the URL.

Prerequisites

Environment	What you need
Hosted (Agentverse)	An Agentverse account. The platform provides uagents. Add requests in the agent’s Python dependencies if it is not already available.
Local / mailbox	Python 3.10+, uagents==0.25.5, and requests.

You also need:

• A structured-output extractor agent on Agentverse (OpenAI, Claude, or equivalent) that accepts StructuredOutputPrompt and replies with StructuredOutputResponse. Paste your extractor address into secrets. Do not hardcode a public demo address; it can go offline or be unreachable from your workspace.
• Optional: a dedicated Solana RPC URL. Public api.mainnet-beta.solana.com is rate-limited (HTTP 429) and is a poor production default. For tests, use devnet as in the on-chain Solana example. For production, use a paid RPC provider.

Chat protocol: publish Agent Chat Protocol (the spec imported from uagents_core.contrib.protocols.chat, typically v0.3.x on Agentverse). See Agent Chat Protocol.

Local install (macOS / Linux)

python3 -m venv .venv
source .venv/bin/activate
pip install "uagents==0.25.5" requests

Local install (Windows)

python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install "uagents==0.25.5" requests

Agentverse secrets

On the hosted agent Build tab, open Agent Secrets and add:

Secret name	Value
AI_AGENT_ADDRESS	Your extractor agent address (agent1q…). Required.
SOLANA_RPC_URL	Optional. Default is public mainnet RPC. Examples: https://api.devnet.solana.com or your provider URL.

The extractor agent must be running, Almanac-registered, and visible to your agent (same workspace or public). Copy its address from the Agentverse profile page.

Overview

The hosted sample uses two files:

File	Role
solana_service.py	Address validation and Solana getBalance RPC
agent.py	Chat protocol + structured-output client

Do not paste both into one file. Create each file in the Agentverse editor.

Message flow

Acknowledgements happen on receipt, before forwarding to the extractor.

1. Query — ASI:One sends a ChatMessage (for example, “What’s the balance of wallet AtTjQKXo1CYTa2MuxPARtr382ZyhPU5YX4wMMpvaa1oy?”).
2. Ack — The Solana agent acknowledges the chat message immediately.
3. Extract — It forwards the text to your extractor agent with SolanaRequest JSON schema.
4. Balance — After a valid address is returned, it calls get_balance_from_address (JSON-RPC getBalance).
5. Reply — It sends a ChatMessage with SOL, lamports, and an Explorer URL.

Implementation (hosted Agentverse)

Create a hosted agent named SolanaWalletAgent. Refer to Hosted Agents.

To add solana_service.py:

1. Click the New File icon.

2. Name the file solana_service.py.

3. Confirm both files are listed separately.

Copy each fence below into its own file. After paste, both files should compile (python -m py_compile solana_service.py agent.py).

1. solana_service.py

RPC URL comes from SOLANA_RPC_URL. Failures raise (they do not return "Error: …" strings), so health checks and chat handlers can treat RPC errors as failures. HTTP runs in a worker thread so the agent event loop is not blocked.

solana_service.py

import asyncio
import json
import logging
import os
import re

import requests
from uagents import Field, Model

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

SOLANA_RPC_URL = os.getenv(
    "SOLANA_RPC_URL",
    "https://api.mainnet-beta.solana.com",
)
LAMPORTS_PER_SOL = 1000000000
BASE58_RE = re.compile(r"^[1-9A-HJ-NP-Za-km-z]{32,44}$")
REQUEST_TIMEOUT_SECONDS = 15

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

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

def is_valid_solana_address(address: str) -> bool:
    if not address or not isinstance(address, str):
        return False
    return BASE58_RE.fullmatch(address.strip()) is not None

def _post_get_balance(address: str) -> dict:
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "getBalance",
        "params": [address],
    }
    headers = {"Content-Type": "application/json"}
    response = requests.post(
        SOLANA_RPC_URL,
        headers=headers,
        json=payload,
        timeout=REQUEST_TIMEOUT_SECONDS,
    )
    response.raise_for_status()
    return response.json()

async def get_balance_from_address(address: str) -> str:
    if not is_valid_solana_address(address):
        raise ValueError(
            "Invalid Solana address. Use a base58 public key (32-44 characters)."
        )

    logger.info("Getting balance for address: %s via %s", address, SOLANA_RPC_URL)

    try:
        result = await asyncio.to_thread(_post_get_balance, address)
    except requests.exceptions.RequestException as exc:
        raise RuntimeError(f"Solana RPC request failed: {exc}") from exc
    except json.JSONDecodeError as exc:
        raise RuntimeError(f"Solana RPC returned invalid JSON: {exc}") from exc

    if "error" in result:
        message = result["error"].get("message", result["error"])
        raise RuntimeError(f"Solana RPC error: {message}")

    value = result.get("result", {}).get("value")
    if value is None:
        raise RuntimeError("No balance information found in RPC response")

    lamports = int(value)
    sol_balance = lamports / LAMPORTS_PER_SOL
    result_str = f"{sol_balance:.9f} SOL ({lamports} lamports)"
    logger.info("Balance for %s: %s", address, result_str)
    return result_str

Public mainnet RPC can return 429 or 403. Switch SOLANA_RPC_URL to devnet while you test, or to a provider URL in production. Explorer links in the chat reply default to mainnet; append ?cluster=devnet when you query devnet.

2. agent.py (hosted)

Hosted Agentverse starts the agent for you. Do not include agent.run(). Set AI_AGENT_ADDRESS as a secret. Beginner sample registers chat and structured-output only.

agent.py

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

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

from solana_service import SolanaRequest, get_balance_from_address, is_valid_solana_address

agent = Agent()

AI_AGENT_ADDRESS = os.getenv("AI_AGENT_ADDRESS", "").strip()
if not AI_AGENT_ADDRESS or AI_AGENT_ADDRESS.startswith("PASTE_"):
    raise ValueError(
        "Set AI_AGENT_ADDRESS to your structured-output extractor agent address."
    )

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]

def _schema(model: type[Model]) -> dict[str, Any]:
    if hasattr(model, "model_json_schema"):
        return model.model_json_schema()
    return model.schema()

def _validate(model: type[Model], data: dict[str, Any]) -> Model:
    if hasattr(model, "model_validate"):
        return model.model_validate(data)
    return model.parse_obj(data)

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

def _sender_key(session_id: Any) -> str:
    return f"chat_sender:{session_id}"

@chat_proto.on_message(ChatMessage)
async def handle_message(ctx: Context, sender: str, msg: ChatMessage):
    ctx.logger.info("Got a message from %s", sender)
    await ctx.send(
        sender,
        ChatAcknowledgement(
            acknowledged_msg_id=msg.msg_id,
            timestamp=datetime.now(timezone.utc),
        ),
    )

    for content in msg.content:
        if isinstance(content, StartSessionContent):
            ctx.logger.info("Got a start session message from %s", sender)
            continue
        if isinstance(content, TextContent):
            ctx.storage.set(_sender_key(ctx.session), sender)
            await ctx.send(
                AI_AGENT_ADDRESS,
                StructuredOutputPrompt(
                    prompt=content.text,
                    output_schema=_schema(SolanaRequest),
                ),
            )
        else:
            ctx.logger.info("Got unexpected content from %s", sender)

@struct_output_client_proto.on_message(StructuredOutputResponse)
async def handle_structured_output_response(
    ctx: Context, sender: str, msg: StructuredOutputResponse
):
    storage_key = _sender_key(ctx.session)
    session_sender = ctx.storage.get(storage_key)
    if session_sender is None:
        ctx.logger.error("Discarding message: no chat sender stored for this session")
        return

    try:
        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

        wallet_request = _validate(SolanaRequest, msg.output)
        address = (wallet_request.address or "").strip()
        if not is_valid_solana_address(address):
            await ctx.send(
                session_sender,
                create_text_chat(
                    "Sorry, I couldn't find a valid Solana wallet address in your query."
                ),
            )
            return

        balance = await get_balance_from_address(address)
        cluster = os.getenv("SOLANA_RPC_URL", "")
        explorer = f"https://explorer.solana.com/address/{address}"
        if "devnet" in cluster:
            explorer = f"{explorer}?cluster=devnet"
        response_text = (
            f"Wallet Balance for `{address}`:\n{balance}\n\n"
            f"[View on Solana Explorer]({explorer})"
        )
        await ctx.send(session_sender, create_text_chat(response_text))
    except Exception as err:
        ctx.logger.error("Error processing structured output: %s", err)
        await ctx.send(
            session_sender,
            create_text_chat(
                "Sorry, I couldn't check the wallet balance. Please try again later."
            ),
        )
    finally:
        ctx.storage.set(storage_key, None)

@chat_proto.on_message(ChatAcknowledgement)
async def handle_ack(ctx: Context, sender: str, msg: ChatAcknowledgement):
    ctx.logger.info(
        "Got an acknowledgement from %s for %s", sender, msg.acknowledged_msg_id
    )

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

Session sender is stored under chat_sender:<session> and cleared after the reply. Hosted storage is shared; do not assume more than one in-flight extractor round-trip per session.

Optional: quota, direct SolanaRequest, and health protocol

These protocols are not required for ASI:One chat. Include them only if you need rate limits, a direct SolanaRequest API, or a health probe.

The health handler awaits get_balance_from_address (no asyncio.run). RPC errors raise, so a bad SOLANA_RPC_URL reports unhealthy. The probe uses the well-known System Program address (valid base58).

optional_protocols.py

from enum import Enum

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

from solana_service import SolanaRequest, SolanaResponse, get_balance_from_address

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("Received wallet balance request for address: %s", msg.address)
    try:
        balance = await get_balance_from_address(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)))

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

SYSTEM_PROGRAM = "11111111111111111111111111111111"

@health_protocol.on_message(HealthCheck, replies={AgentHealth})
async def handle_health_check(ctx: Context, sender: str, msg: HealthCheck):
    status = HealthStatus.UNHEALTHY
    try:
        await get_balance_from_address(SYSTEM_PROGRAM)
        status = HealthStatus.HEALTHY
    except Exception as err:
        ctx.logger.error("Health check failed: %s", err)
    await ctx.send(
        sender,
        AgentHealth(agent_name="solana_wallet_agent", status=status),
    )

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

Merge the optional handlers into agent.py if you use them. Sending HealthCheck while SOLANA_RPC_URL is invalid (for example https://invalid.example) must yield status=unhealthy.

Optional: run locally with mailbox

For a mailbox agent, construct Agent with name, seed, port, and mailbox=True, then start it yourself:

local_run.py

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

See Mailbox Agents. Do not add this block on a hosted Agentverse agent.

Adding a README to your agent

1. Open Overview in the editor.
2. Edit the README so ASI:One can match queries. See Importance of a good README.
3. Confirm Agent Chat Protocol is listed on the profile.

Discoverability checklist

• README states Solana wallet balance, read-only RPC lookup, and example queries.
• Agent is started, Almanac-registered, and chat protocol published.
• Extractor agent (AI_AGENT_ADDRESS) is also running and reachable.
• Test with a query that matches the README, for example: Solana wallet balance for 6wFKPxNToSnggrZr4P4s1r4zRxuJX2nSA7iTdQDPpgHc.

Query your agent from ASI:One

1. Sign in at ASI:One (Google or ASI:One wallet) and start a new chat.
2. Enable the Agents switch so ASI:One can call Agentverse agents.

3. Ask something like: What's the balance of wallet address 6wFKPxNToSnggrZr4P4s1r4zRxuJX2nSA7iTdQDPpgHc.

ASI:One may pick another agent if several match. Prefer a direct chat with your agent’s Agentverse address while you develop, then retry ASI:One with README keywords.

Expected output

For a valid mainnet address you should see a chat reply similar to:

Wallet Balance for `6wFKPxNToSnggrZr4P4s1r4zRxuJX2nSA7iTdQDPpgHc`:
0.000000000 SOL (0 lamports)

[View on Solana Explorer](https://explorer.solana.com/address/6wFKPxNToSnggrZr4P4s1r4zRxuJX2nSA7iTdQDPpgHc)

Zero lamports is a valid RPC result for an unused account. Invalid addresses should fail before RPC with a friendly chat error.

Troubleshooting

Symptom	Likely cause	What to do
Agent hangs after the user message	Extractor unreachable or AI_AGENT_ADDRESS unset	Confirm the secret is the live extractor address; check Almanac and that the extractor is started
ValueError: Set AI_AGENT_ADDRESS	Missing secret	Add AI_AGENT_ADDRESS on Agentverse (or export it locally)
Chat says invalid address	Not base58 / wrong length	Paste a 32–44 character Solana public key
HTTP 429 / 403 from RPC	Public mainnet rate limits	Set SOLANA_RPC_URL to devnet or a paid endpoint
Health always healthy (old sample)	RPC errors returned as strings; asyncio.run did not raise	Use the sample that raises on RPC failure and awaits the probe
parse_obj / .schema() errors	Pydantic v2	Sample uses model_validate / model_json_schema with v1 fallbacks
ASI:One never selects this agent	Weak README / many similar agents	Follow the discoverability checklist; test via Agentverse chat first
Pasted code is one mashed line	Docs copy glitch	Copy from the fences titled solana_service.py and agent.py; python -m py_compile each file
Hosted agent errors on agent.run()	Local run() on a hosted agent	Omit the if __name__ == "__main__" block on Agentverse