agentops-rpc

Shared Pydantic wire types for the AgentOps SDK and controlplane.

Both the SDK (komodor-agentops) and the controlplane import models from this package, giving build-time (mypy) and runtime (Pydantic v2) type safety across the wire boundary.

Install

pip install agentops-rpc

Quick Start

from agentops_rpc import RunEventIn, IngestRunEventsRequest, HOOK_EVENT_TYPE_MAP

event = RunEventIn(
    event_type="span.started",
    span_id="span_abc",
    kind="tool",
    name="Bash",
    hook_event_type="tool.start",
)
event.model_dump()  # JSON-safe dict

request = IngestRunEventsRequest(worker_id="wrk_1", events=[event])

Exported Models

types.py -- Literal type aliases

events.py -- Event ingest models

• RunEventIn -- unified event sent SDK -> controlplane
• TraceEventIn, LogEventIn -- structured events for DB tables
• IngestRunEventsRequest / IngestRunEventsResponse
• BatchEventsRequest
• TraceEventResponse, LogEventResponse, RunMessageResponse

runs.py -- Run lifecycle models

• CreateRunRequest, RunSummary, RunDetail, RunJob
• PollNextRunRequest, ClaimRunResponse
• StartRunRequest, StartRunDirectRequest
• CompleteRunRequest, FailRunRequest, UpdateRunOutputRequest
• InvokeAgentRequest / InvokeAgentResponse
• WorkerInvokeRequest / WorkerInvokeResponse

agents.py -- Agent registration models

• AgentManifest, HeartbeatRequest, HeartbeatResponse
• AgentInstanceResponse

hook_types.py -- Hook event mapping

• HOOK_EVENT_TYPE_MAP -- maps Claude Code HookEventName to HookEventType

a2a.py -- A2A protocol wire types

• A2AMessage, A2AMessagePart
• A2AMessageRequest, A2ATaskResponse

Type Safety Guarantees

Adding a CP↔worker frame type

Four steps, none of them a Python edit. schemas/frames.schema.json is the hand-authored source of truth; Python, Go and TypeScript are all generated from it, and no language is privileged.

1. Add the payload to $defs, and bind it in x-agentops-frames ("my_frame": "MyFramePayload"). Leave additionalProperties out rather than setting it true — the schema's own description explains why the two differ in generated Go.
2. Declare how it travels in x-agentops-delivery — control (send now or fail now) or at-least-once (buffered, retried, may deliver twice, ordered per stream). There is no default and the generator refuses to run without one, because either default is wrong for half the types.
3. Add a fixture under schemas/fixtures/ (and fixtures/invalid/ if the type has a constraint worth proving). A per-type coverage guard fails without one.
4. Run make generate-frames. This writes the Python models and registry, the Go wire types, and the TypeScript bindings together — plus the MY_FRAME constant and the package's __all__, so from agentops_rpc.wire import MY_FRAME, MyFramePayload works with no further edit. make check-frames is the CI guard that the committed artifacts match the schema.

The one thing that does still cost a Python line is a payload with an inline sub-schema: the registry binds top-level payload schemas, so a shape nested inside one is invisible to the generator and each language names it differently. wire/__init__.py aliases it once (see Event as RunEvent). A payload built only from $defs and scalars needs nothing.

Everything after that is feature code: a Router.on(MY_FRAME) handler on the worker side, a caller on the control-plane side. See docs/reference/cp-worker-protocol.md for how frames reach a worker and how a reply is correlated.

Dependencies

Only pydantic>=2.0. Zero internal dependencies.