Use run_traced_async when you want to execute an asynchronous target once and record what it does without running the full AgentIdem fault suite.
It is the async equivalent of run_traced.
Basic usage
import asyncio
from agentidem import run_traced_async
async def main():
trace = await run_traced_async(
"example.async_refund_agent",
async_refund_agent,
)
print(trace)
asyncio.run(main())
The first argument identifies the target.
The second argument is the asynchronous callable AgentIdem should execute.
Example async target
from agentidem import read, write
@read
async def get_payment(payment_id: str):
return await payments.get(payment_id)
@write(identity=lambda payment_id: payment_id)
async def refund(payment_id: str):
return await payments.refund(payment_id)
async def async_refund_agent():
payment = await get_payment("payment-123")
if payment["status"] == "paid":
return await refund("payment-123")
return "no_refund"
Record one traced execution:
import asyncio
from agentidem import run_traced_async
async def main():
trace = await run_traced_async(
"example.async_refund_agent",
async_refund_agent,
)
print(trace)
asyncio.run(main())
What run_traced_async does
run_traced_async executes the target once while AgentIdem records decorated async operations.
Conceptually:
async target starts
↓
await READ get_payment
↓
await WRITE refund
↓
async target returns
↓
trace
It does not run the complete fault suite.
If you want AgentIdem to run baseline execution and fault scenarios against an async target, use test_agent_async.
Trace contents
A trace can contain operation-level information such as:
- operation name
- operation kind
- arguments
- logical identity
- result
- error
- execution status
- observation state
- timestamps
For example:
READ
name: get_payment
status: SUCCESS
observation: RECEIVED
WRITE
name: refund
identity: payment-123
status: SUCCESS
observation: RECEIVED
Async read operations
Async functions decorated with @read appear as READ operations.
from agentidem import read
@read
async def get_order(order_id: str):
return await database.get(order_id)
A traced call may appear conceptually as:
READ
name: get_order
status: SUCCESS
observation: RECEIVED
Reads are recorded for execution context but are not treated as side effects.
Async write operations
Async functions decorated with @write appear as WRITE operations.
from agentidem import write
@write(identity=lambda order_id: order_id)
async def create_order(order_id: str):
return await external_service.create_order(order_id)
A traced call may appear conceptually as:
WRITE
name: create_order
identity: order-123
status: SUCCESS
observation: RECEIVED
Logical identities
If an async write defines an identity:
@write(identity=lambda payment_id: payment_id)
async def refund(payment_id: str):
return await payments.refund(payment_id)
AgentIdem can record the resolved logical identity with the operation.
For example:
operation: refund
identity: payment-123
This can later be used when comparing operations or detecting repeated logical writes.
Running an async target with arguments
If your async target requires arguments, wrap it in another async callable.
import asyncio
from agentidem import run_traced_async
async def target():
return await async_refund_agent("payment-123")
async def main():
trace = await run_traced_async(
"example.async_refund_agent",
target,
)
print(trace)
asyncio.run(main())
This keeps await inside an async function and gives run_traced_async a callable it can execute.
Failed async execution
An asynchronous target may fail after some operations have already been recorded.
For example:
from agentidem import write
@write(identity=lambda order_id: order_id)
async def create_order(order_id: str):
return await external_service.create_order(order_id)
async def failing_agent():
await create_order("order-123")
raise RuntimeError("Unexpected failure")
The write may already have completed before the target raises the exception.
AgentIdem should preserve the partial trace recorded before execution stopped.
TracedExecutionError
When the async target fails during traced execution, AgentIdem can raise:
TracedExecutionError
The error preserves the partial trace recorded before the failure.
This matters because:
async target failed
does not mean:
no side effect occurred
A successful write may already have changed external state.
Inspecting the trace
The returned trace can be inspected directly.
import asyncio
from agentidem import run_traced_async
async def main():
trace = await run_traced_async(
"example.async_refund_agent",
async_refund_agent,
)
print(trace)
asyncio.run(main())
The trace gives you the operation-level execution record rather than the broader reliability report produced by test_agent_async.
Trace serialization
Async traces use the same structured trace model as synchronous traces.
They can be serialized to JSON for:
- debugging
- CI artifacts
- later inspection
- replay
- automated analysis
A saved trace can also be loaded later for replay.
run_traced_async vs test_agent_async
Use run_traced_async when you want:
one async execution
+
structured trace
Use test_agent_async when you want:
baseline execution
+
fault scenarios
+
safety findings
+
structured report
In short:
| API | Purpose |
|---|---|
run_traced_async | Execute an async target once and record the trace |
test_agent_async | Run the AgentIdem reliability test suite against an async target |
Sync targets
run_traced_async is for asynchronous targets.
For synchronous targets, use run_traced.