SUPERVAIZER API Reference

This document provides details about the core components and APIs of SUPERVAIZER.

Core Components

Agent

The Agent class represents a SUPERVAIZE agent and provides methods for agent registration, status checks, and method execution.

from supervaizer import Agent, AgentMethod

agent = Agent(server=server, account=account)
agent.register(name="agent-name")
agent.set_method(AgentMethod.METHOD_NAME, handler_function)

Server

The Server class handles communication with the SUPERVAIZE backend services.

from supervaizer import Server

server = Server(api_url="https://api.example.com")

Account

The Account class manages authentication and user information.

from supervaizer import Account

account = Account(server=server)
account.login(username="username", password="password")

Event-reporting methods are async so asyncio-based agents do not block the event loop:

case = await Case.start(job_id=job_id, account=account, name="Case", description="Work")
await case.update(CaseNodeUpdate(name="Step", payload={"status": "running"}))
await case.close(case_result={"status": "done"})

Synchronous controller methods can use the explicit compatibility shims:

case = Case.start_sync(job_id=job_id, account=account, name="Case", description="Work")
case.update_sync(CaseNodeUpdate(name="Step", payload={"status": "running"}))
case.close_sync(case_result={"status": "done"})

Events

The event system enables communication between components.

from supervaizer import Event, EventType

event = Event.create(EventType.AGENT_REGISTERED, payload={"agent_id": "123"})

Jobs

The job system handles the execution and tracking of agent tasks.

from supervaizer import Job
from supervaizer import EntityStatus

# Check job status
job = agent.get_job(job_id="job-123")
if job.status == EntityStatus.COMPLETED:
    print("Job completed successfully!")

DataResource

Declares a named CRUD data endpoint that the SDK auto-generates as FastAPI routes.

from supervaizer import DataResource, DataResourceField, Editable
from uuid import uuid4

resource = DataResource(
    name="contacts",
    fields=[DataResourceField(name="email", field_type="email", required=True)],
    on_list=lambda: [],
    on_create=lambda data: {**data, "id": str(uuid4())},
)

See Data Resources for full reference.

Editable (enum)

Controls Studio form behaviour for a field.

Value	Meaning
Editable.ALWAYS	Editable on both create and update
Editable.CREATE_ONLY	Writable on create; read-only on edit
Editable.NEVER	Never shown in a form input

Job Polling

The optional job_poll method enables manual polling for agents that depend on external events (webhooks, email services, telephony). When defined, the workbench shows a "Check for updates" button on active jobs.

from supervaizer import AgentMethod

poll_method = AgentMethod(
    name="poll",
    method="my_module.poll_external",
    description="Check external services for updates",
)

The poll handler receives {"job_id": job_id} as kwargs and returns a JobResponse. See Manual Polling for details.

A2A Protocol API

Agent Discovery

Retrieve a list of all available agents:

import requests

response = requests.get("https://your-server/.well-known/agents.json")
agents = response.json()

Agent Cards

Retrieve detailed information about a specific agent:

import requests

response = requests.get(f"https://your-server/.well-known/agents/v1.0.0/myagent_agent.json")
agent_card = response.json()

Health Status

Check the health status of all agents:

import requests

response = requests.get("https://your-server/.well-known/health")
health_data = response.json()

Additional Components

Parameters

The parameter system manages agent-specific configuration:

from supervaizer import Parameter, Parameters

params = Parameters.from_list([
    Parameter(name="api_key", value="your-api-key", is_secret=True),
    Parameter(name="endpoint", value="https://api.example.com")
])

Telemetry

The telemetry system provides monitoring and logging capabilities:

from supervaizer import Telemetry, TelemetryType, TelemetryCategory, TelemetrySeverity

telemetry = Telemetry(
    agent_id="agent-123",
    type=TelemetryType.LOGS,
    category=TelemetryCategory.PERFORMANCE,
    severity=TelemetrySeverity.INFO,
    details={"message": "Processing complete", "duration_ms": 250}
)

Advanced Usage

Creating Custom Agents

Example of creating a custom agent with specialized methods:

from supervaizer import Agent, AgentMethod, AgentMethods

def process_data(data):
    # Process data
    return {"result": "processed"}

# Define agent methods
methods = AgentMethods(
    job_start=AgentMethod(
        name="start_processing",
        method="process_data",
        description="Start processing data"
    ),
    job_stop=AgentMethod(
        name="stop_processing",
        method="stop_process",
        description="Stop processing data"
    ),
    job_status=AgentMethod(
        name="check_status",
        method="get_status",
        description="Check processing status"
    ),
    job_poll=AgentMethod(  # Optional
        name="poll",
        method="poll_external",
        description="Check external services for updates"
    )
)

# Create agent
agent = Agent(
    name="data-processor",
    version="1.0.0",
    description="Agent for processing data",
    methods=methods
)

Uploaded on 2026-04-09 00:25:26