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.

Worse, Agent._handle_message catches every exception a handler raises and writes it to a log. A handler that crashes on every message still looks like a green test. The most common agent bug is invisible to the most common agent 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, not swallowed 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.

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"

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

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