Async Support

Use acall(), async sessions, and async middleware for non-blocking tool execution.

Overview

Every agentguard GuardedTool supports async execution out of the box. Use acall() for async tool invocations, or use the async with tool.session() context manager for session-scoped tracing.

Async Tool Execution

python 
from agentguard import guard

@guard(validate_input=True, max_retries=3)
def search_web(query: str) -> dict:
    return requests.get(f"https://api.search.com?q={{query}}").json()

# Sync call
result = search_web("Python tutorials")

# Async call
result = await search_web.acall("Python tutorials")

Session Context Manager

The async with tool.session() context manager groups multiple tool calls into a single session for trace recording:

python 
from agentguard import guard

@guard(record=True, detect_hallucination=True)
def search_web(query: str) -> dict:
    return requests.get(f"https://api.search.com?q={{query}}").json()

# All calls within the session share a session_id
async with search_web.session() as session:
    r1 = await session.acall("Python tutorials")
    r2 = await session.acall("FastAPI docs")
    # Both calls are recorded under the same session trace

async_record_session()

For programmatic session recording in async contexts:

python 
from agentguard import async_record_session

async with async_record_session(storage="./traces", backend="sqlite", session_id="my-session") as recorder:
    result = await search_web.acall("test query")
    entries = recorder.entries()
    # Traces written to ./traces/agentguard_traces.db

Async Middleware

Middleware functions can be async — the pipeline handles both sync and async middleware transparently:

python 
from agentguard.middleware import MiddlewareChain

# Async middleware
async def auth_middleware(ctx, next):
    if not ctx.metadata.get("api_key"):
        raise PermissionError("No API key")
    return await next(ctx)

# Sync middleware (automatically wrapped)
def logging_middleware(ctx, next):
    print(f"Calling {{ctx.tool_name}}")
    return next(ctx)  # Returns a coroutine, awaited automatically

chain = MiddlewareChain()
chain.use(auth_middleware)
chain.use(logging_middleware)

@guard(middleware=chain)
async def my_tool(x: str) -> str:
    return x.upper()
Mixed sync/async

You can freely mix sync and async middleware in the same chain. Sync middleware is wrapped transparently so that next(ctx) returns a coroutine that the framework awaits.

Async with Integration Wrappers

python 
from agentguard.integrations.crewai_integration import GuardedCrewAITool
from agentguard.integrations.autogen_integration import GuardedAutoGenTool

# CrewAI async
guarded = GuardedCrewAITool(search_web, config=config)
result = await guarded.arun(query="test")

# AutoGen async
guarded = GuardedAutoGenTool(search_web, config=config)
result = await guarded.acall(query="test")
Edit this page on GitHub