veriops-sdk 0.1.1

Python SDK for agent observability event ingestion and runbook validation.

veriops-sdk (Python)

Python SDK for VeriOps Agent Observability - ingest agent execution events and validate runs against YAML runbook specs.

Features

• Event ingestion: Buffered, batched event emission to POST /v1/events with automatic retries
• Context managers: client.run() and run.step() for clean instrumentation
• Runbook validation: Validate runs against YAML specs (allowed tools, required steps, budgets)
• Policy support: Validate using stored policies by ID
• Thread-safe: Safe for concurrent use
• Production-ready: Exponential backoff, error handling, hooks for monitoring

Installation

pip install veriops-sdk

Quick Start

Basic Usage

from veriops_sdk import ObsClient

# Initialize client
client = ObsClient(
    base_url="http://localhost:8000",
    api_key="dev-key",
    project_id="my-project",
)

# Instrument a run with steps
with client.run(runbook="my_runbook_v1") as run:
    with run.step(name="fetch_data", tool="httpx.get", input={"url": "https://api.example.com"}) as step:
        # Your actual work here
        result = httpx.get("https://api.example.com")
        step.set_output({"status_code": result.status_code})
        step.set_tokens_cost(tokens=100, cost_usd=0.001)
    
    run.set_totals(tokens=100, cost_usd=0.001)

# Validate the run
validation = client.validate_run(
    run.run_id,
    runbook_yaml="""
    allowed_tools:
      - httpx.get
    required_steps:
      - name: fetch_data
    budgets:
      max_total_tokens: 1000
      max_total_cost_usd: 1.0
    """
)

print(f"Validation status: {validation['status']}")
print(f"Reasons: {validation['reasons']}")

client.close()

Using Policies

If you've created a policy in the backend, validate using its ID:

validation = client.validate_run(
    run.run_id,
    policy_id="123e4567-e89b-12d3-a456-426614174000"
)

Environment Variables

export OBS_BASE_URL=http://localhost:8000
export OBS_API_KEY=dev-key
export OBS_PROJECT_ID=my-project

import os
from veriops_sdk import ObsClient

client = ObsClient(
    base_url=os.getenv("OBS_BASE_URL", "http://localhost:8000"),
    api_key=os.getenv("OBS_API_KEY", "dev-key"),
    project_id=os.getenv("OBS_PROJECT_ID", "default"),
)

API Reference

ObsClient

Main client for event ingestion and validation.

Constructor

ObsClient(
    base_url: str,
    api_key: str,
    project_id: str,
    *,
    max_batch_events: int = 100,
    flush_interval_events: int = 50,
    timeout_s: float = 10.0,
    max_retries: int = 5,
    backoff_base_s: float = 0.3,
    backoff_cap_s: float = 5.0,
    backoff_jitter: float = 0.2,
    raise_on_flush_error: bool = False,
    on_result: Optional[Callable[[Dict[str, Any]], None]] = None,
    on_error: Optional[Callable[[BaseException], None]] = None,
    http_client: Optional[httpx.Client] = None,
)

Parameters:

• base_url: Backend API base URL (e.g., "http://localhost:8000")
• api_key: API key for authentication
• project_id: Project identifier for grouping runs
• max_batch_events: Maximum events per batch (default: 100)
• flush_interval_events: Auto-flush when buffer reaches this size (default: 50)
• timeout_s: HTTP request timeout in seconds (default: 10.0)
• max_retries: Maximum retry attempts for failed requests (default: 5)
• backoff_base_s: Base backoff delay in seconds (default: 0.3)
• backoff_cap_s: Maximum backoff delay in seconds (default: 5.0)
• backoff_jitter: Jitter factor for backoff (default: 0.2)
• raise_on_flush_error: Raise exception on flush failure (default: False)
• on_result: Optional callback for successful flush results
• on_error: Optional callback for flush errors
• http_client: Optional custom httpx.Client instance

Methods

run(runbook: str) -> RunContext

Create a run context manager. Emits run.start on enter and run.end on exit.

with client.run(runbook="my_runbook") as run:
    # Your agent code here
    run.set_totals(tokens=100, cost_usd=0.01)

step(name: str, tool: str, input: Optional[Dict[str, Any]] = None) -> StepContext

Create a step context manager within a run. Emits step.start on enter and step.end on exit.

with run.step(name="process", tool="my_tool", input={"data": "..."}) as step:
    result = do_work()
    step.set_output({"result": result})
    step.set_tokens_cost(tokens=50, cost_usd=0.005)

validate_run(run_id: str, *, runbook_yaml: Optional[str] = None, policy_id: Optional[str] = None) -> Dict[str, Any]

Validate a run against a runbook spec or policy.

Parameters:

• run_id: UUID of the run to validate
• runbook_yaml: Optional YAML string for the runbook spec
• policy_id: Optional UUID of a policy (takes precedence over runbook_yaml)

Returns: Validation result dict with status, reasons, summary, etc.

flush() -> FlushResult

Manually flush buffered events. Returns a FlushResult with ingestion stats.

close()

Close the HTTP client. Call when done with the client.

RunContext

Context manager for a single agent run.

Attributes

• run_id: str - UUID of the run

Methods

set_totals(*, tokens: Optional[int] = None, cost_usd: Optional[float] = None, **extra: Any)

Set run-level totals (tokens, cost, etc.).

step(name: str, tool: str, input: Optional[Dict[str, Any]] = None) -> StepContext

Create a step within this run.

StepContext

Context manager for a single step within a run.

Methods

set_output(output: Dict[str, Any])

Set the step's output data.

set_tokens_cost(*, tokens: Optional[int] = None, cost_usd: Optional[float] = None)

Set token count and cost for this step.

set_status(status: str)

Set step status (typically "ok" or "error").

Advanced Usage

Error Handling

By default, the SDK logs errors but doesn't raise exceptions. To fail fast:

client = ObsClient(
    base_url="...",
    api_key="...",
    project_id="...",
    raise_on_flush_error=True,  # Raise on flush failures
)

Hooks

Monitor ingestion results and errors:

def on_result(data: Dict[str, Any]) -> None:
    print(f"Ingested: {data.get('ingested')}, Failed: {data.get('failed')}")

def on_error(exc: BaseException) -> None:
    print(f"Error: {exc}")

client = ObsClient(
    base_url="...",
    api_key="...",
    project_id="...",
    on_result=on_result,
    on_error=on_error,
)

Manual Event Emission

For advanced use cases, you can emit events manually:

from veriops_sdk.types import run_start, step_start, step_end, run_end

client.enqueue(run_start(
    run_id="...",
    project_id="...",
    runbook="my_runbook"
))

client.enqueue(step_start(
    run_id="...",
    step_id="...",
    index=0,
    name="my_step",
    tool="my_tool",
    input={"key": "value"}
))

client.flush()

Exception Handling in Steps

Steps automatically capture exceptions:

with run.step(name="risky_operation", tool="my_tool") as step:
    # If this raises, step.end will have status="error"
    # and output will include error details
    risky_function()

Runbook YAML Format

allowed_tools:
  - httpx.get
  - httpx.post
  - openai.chat.completions

required_steps:
  - name: fetch_data
  - name: process

budgets:
  max_total_tokens: 1000
  max_total_cost_usd: 1.0

Examples

See the examples/ directory for complete examples:

• local_demo.py - Basic usage with context managers
• shortify_integration_test.py - Integration with a real API

Requirements

• Python >= 3.9
• httpx >= 0.27.0

License

MIT

Links

• Repository: https://github.com/Ashking07/agent-observability-runbooks
• Backend API: See the main repository for API documentation
• Issues: https://github.com/Ashking07/agent-observability-runbooks/issues