uagent-testkit

Test Fetch.ai uAgents without a network, a wallet, or a sleep().

from uagent_testkit import harness

async def test_echo():
    h = harness(my_agent)
    result = await h.deliver(Ping(text="hi"))

    assert result.reply(Pong).text == "pong:hi"
    assert h.storage["seen"] == 1

Why

The uAgents framework has no testing story. To exercise a handler today you generally have to stand the agent up for real — which means Almanac registration, endpoint resolution, a funded wallet on some paths, and asyncio.sleep() calls sprinkled through the test to wait for messages that may never arrive. That is slow, flaky, and awkward to run in CI.

There is also a subtler problem. Agent._handle_message catches every exception a handler raises and writes it to a log. That is the right call at runtime — an agent processes messages from peers it does not control, and one malformed message must not take the process down. But the same containment applies under pytest, and there is no strict mode to turn it off, so a handler that crashes on every message still looks like a green test.

uagent-testkit routes messages straight to the registered handler with a context that records instead of transmitting. No sockets, no registration, no waiting.

• Handler exceptions are raised rather than contained into a log.
• Storage is in-memory, so state doesn't leak between tests via the JSON file the stock KeyValueStore writes to your working directory.
• Reply contracts are enforced — if @on_message(replies=Pong) doesn't send a Pong, that's an assertion failure instead of a log line nobody reads.
• Interval and lifecycle handlers run on demand, so an @on_interval(period=3600) is testable without waiting an hour.
• Multi-agent conversations replay deterministically, with a loop guard instead of a hung test.
• Agents that read BNB Chain are testable offline — ChainDouble answers BEP-20 and balance queries from state your test sets, and blocks any other outbound HTTP.

Install

pip install uagent-testkit

Requires Python 3.10+ and uagents>=0.22. The pytest fixtures register automatically.

Testing one agent

from uagents import Agent, Context, Model
from uagent_testkit import harness

class Ping(Model):
    text: str

class Pong(Model):
    text: str

agent = Agent(name="echo", seed="echo seed")

@agent.on_message(model=Ping, replies=Pong)
async def on_ping(ctx: Context, sender: str, msg: Ping):
    ctx.storage.set("seen", (ctx.storage.get("seen") or 0) + 1)
    await ctx.send(sender, Pong(text=f"pong:{msg.text}"))

async def test_ping():
    h = harness(agent)
    result = await h.deliver(Ping(text="hi"), sender="agent1qexample")

    result.assert_replied_with(Pong)
    result.assert_reply_contract()          # honoured its replies= declaration
    assert result.reply(Pong).text == "pong:hi"
    assert result.sent[0].destination == "agent1qexample"
    assert h.storage["seen"] == 1

Asserting on failure

async def test_bad_input_is_handled():
    result = await h.deliver(Ping(text=""), raise_errors=False)
    assert isinstance(result.error, ValueError)
    assert "empty" in result.logs.errors[0]

By default deliver() re-raises whatever the handler raised, so a broken handler fails the test loudly. Pass raise_errors=False when the failure is the thing under test.

Sender verification

Handlers without allow_unverified=True are only reachable from a verified agent address. The harness enforces that, so a test can't pass for a flow production rejects:

from uagent_testkit import UnverifiedSender, user_sender

async def test_strict_handler_refuses_users():
    with pytest.raises(UnverifiedSender):
        await h.deliver(Ping(text="hi"), sender=user_sender())

Intervals and lifecycle

await h.startup()                  # runs @on_event("startup") handlers
result = await h.tick()            # runs @on_interval handlers once, ignoring period
result = await h.tick(only="heartbeat")   # just one, by function name
await h.shutdown()

Testing agents against each other

from uagent_testkit import AgentNetwork

async def test_conversation():
    net = AgentNetwork(alice, bob)
    transcript = await net.send(Ping(text="hello"), to=bob, sender=alice)

    transcript.assert_delivered(Pong, to=alice.address)
    assert transcript.of(Pong)[0].text == "pong:hello"
    assert net.harness(alice).storage["last_pong"] == "pong:hello"

Every reply addressed to another agent in the network is delivered in turn until the conversation goes quiet. Messages sent to addresses outside the network are collected on transcript.undelivered rather than silently dropped:

transcript.assert_all_delivered()   # fails if anything went to an unknown address

Agents that answer each other forever raise ConversationTooLong after max_rounds deliveries instead of hanging the suite. This catches echo loops, which are easy to write and expensive to discover in production:

# a keyword responder whose answer contains its own trigger word
if "menu" in msg.text().lower():
    await ctx.send(sender, chat("here is the menu"))   # -> contains "menu"

Two agents like that will talk to each other indefinitely. On a live network that is two agents spamming each other; here it fails in milliseconds with the transcript so far attached.

Protocols and the chat protocol

Agents assembled from Protocol objects work the same way, including the shared chat protocol from uagents_core that Agentverse and Agent Launch agents use:

chat = Protocol(spec=chat_protocol_spec)

@chat.on_message(ChatMessage)
async def on_chat(ctx: Context, sender: str, msg: ChatMessage):
    ...

agent.include(chat)

async def test_chat():
    h = harness(agent)
    result = await h.deliver(
        ChatMessage(
            timestamp=datetime.now(timezone.utc),
            msg_id=uuid4(),
            content=[TextContent(type="text", text="show me the menu")],
        )
    )
    assert result.replies(ChatAcknowledgement)
    assert result.replies(ChatMessage)[0].text() == "here is what we have"

Agents that read BNB Chain

The harness intercepts ctx.send, which covers agent-to-agent messaging. It does not cover a handler that reaches out to an RPC endpoint — an agent reading a BEP-20 balance still talks to the network during a test. ChainDouble closes that gap:

from uagent_testkit import ChainDouble, FET_BSC, harness

async def test_treasury_agent():
    chain = ChainDouble()
    chain.add_token(FET_BSC, symbol="FET", decimals=18,
                    balances={wallet: 5_500_000_000_000_000_000})

    h = harness(treasury_agent)
    with chain.install():
        result = await h.deliver(CheckBalance(wallet=wallet))

    assert result.reply(BalanceReport).amount == 5.5
    chain.assert_called("eth_call")

It patches whichever of requests, httpx and aiohttp are importable, so it works whether the agent uses web3.py or raw async HTTP, and it understands web3.py's JSON-encoded data= body as well as json=.

Supported: eth_call (BEP-20 balanceOf, symbol, name, decimals, totalSupply), eth_getBalance, eth_getCode, eth_blockNumber, eth_chainId, batched requests.

Two deliberate refusals, both of which turn a silent pass into a loud failure:

# reading state the test never set up
with pytest.raises(UnstubbedCall):
    ...                       # rather than quietly returning zero

# any non-JSON-RPC HTTP from inside a handler
with pytest.raises(NetworkCallBlocked):
    ...                       # rather than hitting a real API in CI

There's also a bsc fixture that gives you an installed double directly.

pytest fixtures

Installing the package registers two factory fixtures:

def test_with_fixture(agent_harness, agent_network):
    h = agent_harness(my_agent)
    net = agent_network(alice, bob)

API

harness(agent)	Wrap an agent. Swaps in in-memory storage and merges the internal protocol.
h.deliver(msg, sender=…, raise_errors=True)	Route a message to its handler. Returns Delivery.
h.tick(only=None)	Run @on_interval handlers once.
h.startup() / h.shutdown()	Run @on_event handlers.
h.storage	In-memory store, subscriptable.
h.handles(Model)	Whether a handler is registered for a type.
Delivery.sent	List of Sent(destination, schema_digest, body).
Delivery.reply(Model)	The single reply of that type; raises otherwise.
Delivery.replies(Model, to=…)	Replies of that type to the sender; to=ANY for any destination.
user_sender()	A user (non-agent) address, for allow_unverified paths.
Delivery.error / .logs	Handler exception and captured log records.
Delivery.assert_replied_with / assert_silent / assert_no_errors / assert_reply_contract	Chainable assertions.
AgentNetwork(*agents)	Multi-agent network.
net.send(msg, to=…, sender=…, max_rounds=20)	Play out a conversation. Returns Transcript.
Transcript.of(Model) / .to(addr) / .undelivered	Inspect what was exchanged.

Scope

This exercises your handler logic: routing, replies, storage, protocol contracts, and inter-agent flow. It deliberately does not simulate Almanac resolution, envelope signing, or on-chain settlement — those are integration concerns and mocking them would give you false confidence. Keep a thin integration test on a real testnet for those, and use this for everything else.

Running the tests

python -m venv .venv && .venv/bin/pip install -e ".[test]"
.venv/bin/python -m pytest

License

Apache-2.0