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 checked — if @on_message(replies=Pong) sends neither a Pong nor an ErrorMessage, assert_reply_contract() fails, and the error the framework would log lands in result.logs, where assert_no_errors() sees it.
• 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 request made through requests, httpx, aiohttp or urllib.

Install

pip install uagent-testkit

Requires Python 3.10+ and uagents>=0.23. 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 str(result.error)

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. Cancellation and KeyboardInterrupt are not handler failures, so they always propagate.

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
                                   # (and @on_schedule ones, on uagents 0.26+)
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. As on a live network, a reply carries the session of the message it answers, so the whole conversation runs in one ctx.session (pass session= to choose it). Sending to an agent you never added raises KeyError. Messages a handler sends 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 requests, httpx and aiohttp where they are importable, plus urllib, at the call every request through them passes, so it works whether the agent uses web3.py or raw HTTP; the test suite drives web3.py's HTTPProvider and AsyncHTTPProvider through it. It understands web3.py's JSON-encoded data= body as well as json=. It does not cover http.client or urllib3 used directly, or sockets a handler opens itself.

Supported: eth_call (BEP-20 balanceOf, symbol, name, decimals, totalSupply), eth_getBalance, eth_getCode, eth_blockNumber, eth_chainId, net_version, eth_getTransactionCount, batched requests. Two of these answer without being told: eth_getCode returns 0x (no code) for any address not registered with add_token(), and eth_getTransactionCount always returns 0.

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 other HTTP from inside a handler: a price API, an LLM, a webhook
with pytest.raises(NetworkCallBlocked):
    ...                       # rather than hitting a real API in CI

Agents wrap RPC reads in try/except so one bad call can't crash them, and that habit would swallow a refusal and turn it back into a silent pass. So deliver() (and tick(), startup(), shutdown()) re-raises any refusal the handler's own work made, caught or not: in the handler, in the tasks it starts, and in asyncio.to_thread(). Work the handler did not start is not charged to it, such as a delivery running alongside it or the manifest upload that include(proto, publish_manifest=True) queues. A refusal from a thread that carries no context (loop.run_in_executor(), a plain Thread) fails every delivery running at that moment. chain.refusals keeps every refusal either way.

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 (and @on_schedule) 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 (or a refusal it caught) 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, session=None)	Play out a conversation in one session. 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.

ctx.send_and_receive() is not supported. Sends are recorded rather than delivered, so no answer can reach the handler that is waiting for one; the harness raises NotImplementedError rather than hand it None, and a handler that catches it still fails its delivery.

The harness can't stop work an agent schedules on its own event loop. include(proto, publish_manifest=True) queues an upload of the protocol's manifest to Agentverse, and it runs whenever that loop runs, with or without the harness: for an agent built inside an async test, as soon as the test yields to the loop. Pass publish_manifest=False when you build agents for tests.

The changelog lists what each release fixed, and which fixes can fail a test that used to pass.

Running the tests

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

License

Apache-2.0