r3fresh 1.0.0

Agent Lifecycle Management SDK - Track AI agent execution with policy enforcement, event emission, and structured events for downstream analytics

r3fresh ALM SDK

Agent Lifecycle Management SDK – A production-ready Python SDK for tracking AI agent execution with policy enforcement, event emission, and structured event data for downstream analytics.

Overview

The SDK provides automatic instrumentation for AI agents, capturing:

• Tool calls with policy enforcement (allow/deny) and latency tracking
• Run lifecycle (run.start / run.end) with summary statistics
• Tasks (task.start / task.end) for logical units of work
• Handoffs for agent-to-agent transitions
• Structured errors (type, message, source, retryable) in tool and run events
• Version tracking (schema, SDK, agent, policy) on every event

All events are emitted automatically. They can be sent to stdout (development) or an HTTP endpoint (production). The SDK does not perform analytics itself; it produces events for your backend or analytics pipeline.

Getting Started

To use r3fresh in production with cloud analytics:

1. Sign up at r3fresh.dev
2. Get your API key from the Dashboard
3. Install the SDK: pip install r3fresh
4. Integrate using the examples below

The r3fresh platform provides:

• Real-time monitoring dashboard for all your agents
• Analytics with run history, tool usage, and performance metrics
• Code Assistant - AI-powered tool to automatically integrate r3fresh into your existing code
• Policy management and quota tracking

Installation

pip install r3fresh

Or install from source:

git clone https://github.com/r3fresh-alm/r3fresh.git
cd r3fresh
pip install -e .

Quick Start

from r3fresh import ALM

# Initialize the SDK
alm = ALM(
    agent_id="my-agent",
    env="development",
    mode="stdout",  # or "http" with endpoint (base URL)
    agent_version="1.0.0",
)

# Define tools with automatic policy enforcement
@alm.tool("search_web")
def search_web(query: str) -> str:
    """Search the web for information."""
    # Your tool implementation
    return f"Results for: {query}"

# Run your agent with automatic tracking
with alm.run(purpose="Process user query"):
    result = search_web("Python SDK documentation")
    print(result)

# All events are automatically captured and emitted

Core Concepts

ALM Instance

The ALM class is the main entry point that manages:

• Event collection and emission
• Policy enforcement
• Run tracking
• Version management

alm = ALM(
    agent_id="unique-agent-id",       # Required: Unique identifier
    env="production",                 # Environment name
    mode="http",                      # "stdout" or "http"
    endpoint="https://api.r3fresh.dev",  # r3fresh platform API (or your own backend)
    api_key="your-api-key",           # Get from https://r3fresh.dev/dashboard
    agent_version="1.2.3",            # Optional: Agent version
    policy_version="2.0.0",           # Optional: Policy version
    allowed_tools={"tool1", "tool2"}, # Optional: Whitelist tools
    denied_tools={"dangerous_tool"},  # Optional: Blacklist tools
    default_allow=True,               # Allow tools by default
    max_tool_calls_per_run=100,       # Optional: Budget limit
)

Runs

Runs track the execution lifecycle of an agent. All events within a run are correlated via run_id.

with alm.run(purpose="Answer user question"):
    # Your agent logic here
    pass
# run.end event automatically emitted with summary statistics

Tools

Tools are automatically instrumented with:

• Policy enforcement (allow/deny)
• Latency tracking (policy_latency_ms, tool_latency_ms, total_latency_ms)
• Structured errors (type, message, source, retryable) on failure or deny
• attempt and retries in events (retry infrastructure exists but retries are disabled by default)

@alm.tool("my_tool")  # Optional: specify tool name
def my_tool(param1: str, param2: int) -> dict:
    """Tool documentation."""
    # Your implementation
    return {"result": "success"}

Tasks

Tasks represent logical units of work within a run. They automatically emit task.start and task.end events.

with alm.task(description="Process user input"):
    # Task logic
    pass
# Success/failure automatically tracked

Handoffs

Handoffs represent agent-to-agent transitions:

alm.handoff(
    to_agent_id="next-agent",
    reason="Requires specialized knowledge",
    context={"query": "..."}
)

Features

Automatic Event Tracking

The SDK automatically captures:

• Run lifecycle: run.start, run.end (with summary stats in run.end metadata)
• Tool execution: tool.request, policy.decision, tool.response (including status="denied" when blocked)
• Task tracking: task.start, task.end
• Handoffs: handoff events

Every event includes a unique event_id (UUID) for idempotency and deduplication.

Policy Enforcement

Enforce tool usage policies:

• Whitelist/blacklist tools
• Budget limits per run
• Default allow/deny behavior

alm = ALM(
    agent_id="agent-1",
    allowed_tools={"safe_tool", "read_tool"},  # Only these allowed
    denied_tools={"delete_tool"},              # Never allow this
    max_tool_calls_per_run=50                  # Budget limit
)

Structured Error Handling

Errors in tool.response and run.end are structured as:

• type, message, source (tool, policy, agent, system)
• retryable (auto-detected for e.g. ConnectionError, TimeoutError, or when "timeout" appears in the message)
• code (optional)

Example (in tool.response metadata or run.end metadata):

{
    "error": {
        "type": "ConnectionError",
        "message": "Connection timeout",
        "source": "tool",
        "retryable": true
    }
}

Version Tracking

All events include version information for drift detection:

• schema_version: Event schema version
• sdk_version: SDK version
• agent_version: Your agent version
• policy_version: Your policy version

Run Summary Statistics

Every run.end event includes metadata.summary:

{
    "metadata": {
        "success": true,
        "summary": {
            "tool_calls": { "total": 10, "allowed": 8, "denied": 1, "error": 1, "retried": 2 },
            "latencies": { "avg_tool_ms": 50.2, "avg_policy_ms": 0.5, "total_run_ms": 1500.0 },
            "tasks": { "completed": 3, "failed": 1 },
            "handoffs": 1
        }
    }
}

On run failure, metadata.error contains the structured error object.

Retries

The SDK records attempt and retries on tool events and marks errors as retryable when appropriate. Automatic retries are not enabled by default (max_retries=0). The infrastructure is in place for future use or custom retry logic.

API Reference

ALM Class

__init__(...)

Initialize an ALM instance.

Parameters:

• agent_id (str, required): Unique agent identifier
• env (str, default="development"): Environment name
• mode (str, default="stdout"): Event sink mode ("stdout" or "http")
• endpoint (str, optional): Base URL for HTTP mode (required if mode="http"). The SDK POSTs to /v1/events. For the r3fresh platform, use https://api.r3fresh.dev.
• api_key (str, optional): API key for authentication. Get yours at r3fresh.dev/dashboard. Sent as Authorization: Bearer <api_key>.
• agent_version (str, optional): Agent version string
• policy_version (str, optional): Policy version string
• allowed_tools (Set[str], optional): Whitelist of allowed tools
• denied_tools (Set[str], optional): Blacklist of denied tools
• default_allow (bool, default=True): Allow tools by default
• max_tool_calls_per_run (int, optional): Maximum tool calls per run

run(purpose: Optional[str] = None) -> Run

Create a run context manager. Returns a Run instance that tracks all events within the run.

tool(tool_name: Optional[str] = None)

Decorator factory for wrapping tool functions with automatic instrumentation.

task(task_type: Optional[str] = None, description: Optional[str] = None) -> TaskContext

Create a task context manager for tracking task execution.

handoff(to_agent_id: str, reason: Optional[str] = None, context: Optional[Dict] = None)

Emit a handoff event for agent-to-agent transitions.

flush()

Flush queued events to the configured sink.

Examples

Basic Agent with Tools

from r3fresh import ALM

alm = ALM(
    agent_id="research-agent",
    env="production",
    mode="http",
    endpoint="https://api.r3fresh.dev",
    api_key="your-api-key",
    agent_version="1.0.0",
)

@alm.tool("search")
def search(query: str) -> str:
    """Search the web."""
    # Implementation
    return "results"

@alm.tool("summarize")
def summarize(text: str) -> str:
    """Summarize text."""
    # Implementation
    return "summary"

with alm.run(purpose="Research and summarize"):
    results = search("Python SDK")
    summary = summarize(results)
    print(summary)

Policy Enforcement

alm = ALM(
    agent_id="controlled-agent",
    denied_tools={"delete", "modify"},
    allowed_tools={"read", "search"},
    max_tool_calls_per_run=20,
)

@alm.tool("delete")
def delete_item(item_id: str):
    """This will be denied."""
    pass

with alm.run():
    try:
        delete_item("123")  # Raises PermissionError
    except PermissionError as e:
        print(f"Blocked: {e}")

Task Management

alm = ALM(agent_id="task-agent", mode="stdout")

with alm.run():
    with alm.task(description="Process input"):
        # Task logic
        pass
    
    with alm.task(description="Generate output"):
        # Task logic
        pass

Agent Handoffs

alm = ALM(agent_id="coordinator", mode="stdout")

with alm.run():
    # Do some work
    if needs_specialist:
        alm.handoff(
            to_agent_id="specialist-agent",
            reason="Requires domain expertise",
            context={"query": user_query}
        )

Error Handling

alm = ALM(agent_id="api-agent", mode="stdout")

@alm.tool("api_call")
def api_call(url: str):
    """API call that may fail."""
    response = httpx.get(url, timeout=5.0)
    response.raise_for_status()
    return response.json()

with alm.run():
    try:
        result = api_call("https://api.example.com/data")
    except Exception:
        # SDK already emitted tool.response with status="error" and structured error
        pass

Event Schema

All events share a common shape. Timestamps are RFC3339 (e.g. 2026-01-21T12:00:00.123Z).

{
    "event_id": "550e8400-e29b-41d4-a716-446655440000",
    "timestamp": "2026-01-21T12:00:00.123Z",
    "event_type": "tool.request",
    "agent_id": "agent-123",
    "env": "production",
    "run_id": "run-456",
    "schema_version": "1.0",
    "sdk_version": "0.1.0",
    "agent_version": "1.0.0",
    "policy_version": "2.0.0",
    "metadata": {}
}

event_id is a UUID per event for idempotency and deduplication.

Event Types

• run.start: Run started
• run.end: Run finished (includes metadata.summary and optionally metadata.error)
• tool.request: Tool call initiated (metadata includes tool_name, tool_call_id, args, etc.)
• policy.decision: Allow or deny (metadata.decision, metadata.tool_call_id, metadata.latency_ms)
• tool.response: Tool completed (metadata.status: success, denied, or error; latencies, attempt, retries)
• task.start: Task started
• task.end: Task finished (success/failure, optional metadata.error)
• handoff: Agent-to-agent handoff

Development Mode

For development, use mode="stdout" to see events as JSON lines:

alm = ALM(agent_id="dev-agent", mode="stdout")

with alm.run():
    @alm.tool("test_tool")
    def test_tool(x: int) -> int:
        return x * 2
    
    result = test_tool(5)

Output:

{"event_type": "run.start", ...}
{"event_type": "tool.request", ...}
{"event_type": "tool.response", ...}
{"event_type": "run.end", ...}

Production Mode

For production with the r3fresh platform:

alm = ALM(
    agent_id="prod-agent",
    mode="http",
    endpoint="https://api.r3fresh.dev",  # r3fresh platform API
    api_key=os.getenv("ALM_API_KEY"),    # Get from https://r3fresh.dev/dashboard
    agent_version=__version__,
)

Events are batched (default 50) and POSTed to {endpoint}/v1/events. The SDK:

• Buffers events and flushes on batch size or at run end
• Catches flush failures so the agent does not crash
• Sends Authorization: Bearer <api_key> when api_key is set

Self-hosted option: You can also run your own event ingestion API. The SDK will POST events to any endpoint that accepts the r3fresh event schema at /v1/events.

Testing

Run the test suite:

pytest

Run the example:

python examples/toy_agent.py

Requirements

• Python 3.8+
• pydantic
• httpx

Resources

• Website: r3fresh.dev
• Documentation: r3fresh.dev/docs
• Dashboard: r3fresh.dev/dashboard
• Code Assistant: r3fresh.dev/dashboard/code-assistant - AI-powered integration helper
• GitHub: github.com/r3fresh-alm/r3fresh

License

r3fresh is distributed under the terms of the MIT license.

Contributing

Contributions are welcome! Please open an issue or submit a pull request.

Support

For issues, questions, or feature requests:

• Email: [e-mail]
• GitHub Issues: github.com/r3fresh-alm/r3fresh/issues