Prefactor HTTP Client

A low-level async HTTP client for the Prefactor API.

Features

• Typed Endpoint Clients: Dedicated clients for agent instances, agent spans, and bulk operations
• Automatic Retries: Exponential backoff with jitter for transient failures
• Type Safety: Full Pydantic models for all request/response data
• Clear Error Hierarchy: Specific exception types for different failure modes
• Idempotency: Built-in support for idempotency keys

Installation

pip install prefactor-http

Quick Start

import asyncio
from prefactor_http import PrefactorHttpClient, HttpClientConfig


async def main():
    config = HttpClientConfig(
        api_url="https://app.prefactorai.com",
        api_token="your-api-token",
    )

    async with PrefactorHttpClient(config) as client:
        instance = await client.agent_instances.register(
            agent_id="agent_123",
            agent_version={"name": "My Agent", "external_identifier": "v1.0.0"},
            agent_schema_version={
                "external_identifier": "v1.0.0",
                "span_type_schemas": [
                    {
                        "name": "agent:llm",
                        "title": "LLM Call",
                        "description": "A call to a language model",
                        "params_schema": {
                            "type": "object",
                            "properties": {
                                "model": {"type": "string"},
                                "prompt": {"type": "string"},
                            },
                            "required": ["model", "prompt"],
                        },
                        "result_schema": {
                            "type": "object",
                            "properties": {"response": {"type": "string"}},
                        },
                        "template": "{{model}}: {{prompt}} → {{response}}",
                    },
                ],
            },
        )
        print(f"Registered instance: {instance.id}")


asyncio.run(main())

agent_instances.register() supports two auth modes:

• Account-scoped token: pass agent_id and usually environment_id.
• Deployment-scoped token: omit agent_id and environment_id; the API derives both from the token.

Endpoints

Agent Instances (client.agent_instances)

# Register a new agent instance
instance = await client.agent_instances.register(
    agent_id="agent_123",
    agent_version={
        "name": "My Agent",
        "external_identifier": "v1.0.0",
        "description": "Optional description",
    },
    agent_schema_version={
        "external_identifier": "schema-v1",
        "span_type_schemas": [
            {
                "name": "agent:llm",
                "title": "LLM Call",  # Optional
                "description": "A call to a language model",  # Optional
                "params_schema": {"type": "object", "properties": {...}},
                "result_schema": {"type": "object", "properties": {...}},  # Optional
                "template": "{{model}}: {{prompt}} → {{response}}",  # Optional
            },
        ],
        # Alternatively, use flat maps for simpler cases:
        # "span_schemas": {"agent:llm": {"type": "object", ...}},
        # "span_result_schemas": {"agent:llm": {"type": "object", ...}},
    },
    id=None,  # Optional: pre-assign an ID
    idempotency_key=None,  # Optional: idempotency key
    update_current_version=True,  # Optional: update the agent's current version
)

# Register with a deployment-scoped token
instance = await client.agent_instances.register(
    agent_version={
        "name": "My Agent",
        "external_identifier": "v1.0.0",
    },
    agent_schema_version={
        "external_identifier": "schema-v1",
        "span_type_schemas": [],
    },
)

# Start an instance
instance = await client.agent_instances.start(
    agent_instance_id=instance.id,
    timestamp=None,  # Optional: override start time
    idempotency_key=None,
)

# Finish an instance
instance = await client.agent_instances.finish(
    agent_instance_id=instance.id,
    status=None,  # Optional: "complete" | "failed" | "cancelled"
    timestamp=None,  # Optional: override finish time
    idempotency_key=None,
)

The AgentInstance response includes: id, agent_id, status, started_at, finished_at, span_counts, and more.

Agent Spans (client.agent_spans)

# Create a span
span = await client.agent_spans.create(
    agent_instance_id="instance_123",
    schema_name="agent:llm",
    status="active",
    payload={"model": "gpt-4", "prompt": "Hello"},  # Optional
    result_payload=None,  # Optional
    id=None,  # Optional: pre-assign an ID
    parent_span_id=None,  # Optional: parent for nesting
    started_at=None,  # Optional: override start time
    finished_at=None,
    idempotency_key=None,
)

# Finish a span
span = await client.agent_spans.finish(
    agent_span_id=span.id,
    status=None,  # Optional: "complete" | "failed" | "cancelled"
    result_payload=None,  # Optional: final result data
    timestamp=None,  # Optional: override finish time
    idempotency_key=None,
)

The AgentSpan response includes: id, agent_instance_id, schema_name, status, payload, result_payload, parent_span_id, started_at, finished_at, and more.

Bulk Operations (client.bulk)

Execute multiple POST actions in a single HTTP request.

from prefactor_http import BulkRequest, BulkItem

request = BulkRequest(
    items=[
        BulkItem(
            _type="agent_instances/register",
            idempotency_key="register-instance-001",
            agent_id="agent_123",
            agent_version={"name": "My Agent", "external_identifier": "v1.0.0"},
            agent_schema_version={
                "external_identifier": "v1.0.0",
                "span_type_schemas": [
                    {
                        "name": "agent:llm",
                        "title": "LLM Call",
                        "params_schema": {
                            "type": "object",
                            "properties": {
                                "model": {"type": "string"},
                                "prompt": {"type": "string"},
                            },
                            "required": ["model", "prompt"],
                        },
                        "result_schema": {
                            "type": "object",
                            "properties": {"response": {"type": "string"}},
                        },
                    },
                ],
            },
        ),
        BulkItem(
            _type="agent_spans/create",
            idempotency_key="create-span-001",
            agent_instance_id="instance_123",
            schema_name="agent:llm",
            status="active",
        ),
    ]
)

response = await client.bulk.execute(request)

for key, output in response.outputs.items():
    print(f"{key}: {output.status}")  # "success" or "error"

Validation rules:

• Each item must have a unique idempotency_key (8–64 characters)
• The request must contain at least one item

Error Handling

from prefactor_http import (
    PrefactorHttpError,
    PrefactorApiError,
    PrefactorAuthError,
    PrefactorNotFoundError,
    PrefactorValidationError,
    PrefactorRetryExhaustedError,
    PrefactorClientError,
)

try:
    async with PrefactorHttpClient(config) as client:
        instance = await client.agent_instances.register(...)
except PrefactorValidationError as e:
    print(f"Validation error: {e.errors}")
except PrefactorAuthError:
    print("Authentication failed - check your API token")
except PrefactorNotFoundError:
    print("Resource not found")
except PrefactorRetryExhaustedError as e:
    print(f"Request failed after retries: {e.last_error}")
except PrefactorApiError as e:
    print(f"API error {e.status_code}: {e.code}")

Configuration

config = HttpClientConfig(
    # Required
    api_url="https://app.prefactorai.com",
    api_token="your-token",
    # Retry behavior
    max_retries=3,
    initial_retry_delay=1.0,
    max_retry_delay=60.0,
    retry_multiplier=2.0,
    # Timeouts
    request_timeout=30.0,
    connect_timeout=10.0,
)

Types

from prefactor_http import AgentStatus, FinishStatus

# AgentStatus = Literal["pending", "active", "complete", "failed", "cancelled", "terminated"]
# FinishStatus = Literal["complete", "failed", "cancelled"]

License

MIT