Quick start

The AgentIdem Nebutex SDK lets you mark state-observing operations as reads, mark state-changing operations as writes, and test how your agent behaves under retry and failure conditions.

Install

If you have not installed the package yet:

pip install agentidem

Define a read operation

Use @read for operations that observe state without changing it.

from agentidem import read

@read
def get_order(order_id: str):
    return database.get(order_id)

Read operations appear in traces, but they are not treated as side effects.

Define a write operation

Use @write for operations that can change external state.

from agentidem import write

@write
def create_order(order_id: str):
    return external_service.create_order(order_id)

Writes can also define a logical identity.

from agentidem import write

@write(identity=lambda payment_id, amount: payment_id)
def charge(payment_id: str, amount: int):
    return payments.charge(payment_id, amount)

The identity tells AgentIdem what makes repeated writes represent the same logical side effect.

Define your agent

Your agent can call the decorated operations normally.

def run_agent(order_id: str):
    order = get_order(order_id)

    if order is None:
        return create_order(order_id)

    return order

Run a reliability test

Use test_agent to run the synchronous AgentIdem test suite.

from agentidem import test_agent

report = test_agent(
    "example.order_agent",
    lambda: run_agent("order-123"),
)

The test suite can run a baseline execution and controlled fault scenarios against the agent.

Inspect the report

The returned report contains structured information about the test run.

Depending on the execution, it can include:

baseline result
fault count
unsafe count
individual fault results
findings
invariant results
safety status

For example:

print(report)

Run a traced execution

If you only want to record what an agent does without running the full fault suite, use run_traced.

from agentidem import run_traced

trace = run_traced(
    "example.order_agent",
    lambda: run_agent("order-123"),
)

The trace records the read and write operations performed during execution.

Async agents

For asynchronous agents, use test_agent_async and run_traced_async.

For example:

import asyncio

from agentidem import test_agent_async

async def main():
    report = await test_agent_async(
        "example.async_agent",
        async_agent,
    )

    print(report)

asyncio.run(main())

For traced async execution:

import asyncio

from agentidem import run_traced_async

async def main():
    trace = await run_traced_async(
        "example.async_agent",
        async_agent,
    )

    print(trace)

asyncio.run(main())