komodor-agentops

Python SDK for AgentOps observability, including the worker runtime, with optional extras for framework integrations (LangChain, Claude Agent SDK, ADK, Agno).

Install

pip install komodor-agentops

# With framework adapters
pip install komodor-agentops[langchain]
pip install komodor-agentops[claude-code]
pip install komodor-agentops[adk]
pip install komodor-agentops[agno]
pip install komodor-agentops[all]           # Everything

Quick Start

@observe() decorator

Wrap functions to emit span events automatically:

from komodor_agentops import observe

@observe(name="summarize", as_type="llm")
async def summarize(text: str) -> str:
    ...

AgentOps client (lightweight event buffer)

from komodor_agentops import AgentOps

client = AgentOps(agent_id="my-agent", endpoint="http://localhost:8000")
await client.log("Processing started", run_id="run_1")
await client.flush()

AgentOpsWorker — the whole of a worker's main()

A worker takes runs and returns results. It serves no HTTP and needs no inbound network access: it connects out to the control plane, is told when work is available, and claims it.

from pathlib import Path

from komodor_agentops import AgentOpsWorker, AgentSpec, Run

spec = AgentSpec.from_dir(
    Path(__file__).parent,
    agent_card={"skills": [{"id": "search", "name": "search"}]},
)

async def on_run(run: Run) -> dict:
    return {"answer": f"handled {run.input.get('prompt', '')}"}

def main() -> None:
    AgentOpsWorker(agent=spec, on_run=on_run).run()

run.input is the payload the run was created with; the dict you return becomes the run's output. Raise to fail the run — that is on_run's only failure channel, and the error is recorded on the run.

No host or port. Nothing dials your worker, so there is nothing to bind. A deployment needs egress to the control plane and no Service, Ingress or open port; the customer namespace chart denies inbound traffic by default.

A worker built against komodor-agentops 0.1.x may instead be using run_worker(agent, handler) with a handler taking an A2AMessage and returning an A2ATask, served on a port. That runtime is gone: it served an A2A app nothing dials any more. Move the handler to on_run= — it receives a Run and returns a mapping — and construct AgentOpsWorker from the package root as above. Run.input is the whole payload the message's parts used to carry, and the answer's human-readable half is the reserved text key rather than an artifact.

The standard AgentOps agent layout is:

my_agent/
  agent-spec.yaml
  agent.md
  worker.py
  skills/
    triage.md
    rca/SKILL.md

AgentSpec.from_dir(Path(__file__).parent, ...) requires agent-spec.yaml, then loads agent.md and skills/ automatically. The worker registers loaded skills during heartbeat, and SDK-owned LLM adapters can prepend the agent context at invocation time.

schema_version: 1
agent_id: my-agent
name: My Agent
description: Does useful work.
owner: AgentOps
repo: https://github.com/komodorio/agentops
source_path: packages/workers/my_agent
labels:
  category: example

agent-spec.yaml can also declare triggers — synced to the control plane on heartbeat and shown on the Fleet → Triggers surface. Supported types are schedule (cron), webhook (inbound HTTP endpoint), and slack_channel (subscribe the agent to Slack channels by NAME, not ID — the control plane materialises a routing rule per channel on heartbeat and dispatches matching messages; the agent is invoked when @mentioned in one of those channels):

triggers:
  - id: incidents-sub
    type: slack_channel
    name: Incident channels
    channels:
      - "#incidents" # normalized to "incidents" — lowercase, no "#"
      - alerts

To make a worker available in the AgentOps Chat UI, advertise chat capability on its agent card:

spec = AgentSpec(
    agent_id="my-chat-agent",
    agent_card={
        "name": "My Chat Agent",
        "capabilities": {"chat": True, "streaming": True},
    },
)

A chat run arrives like any other run: run.input carries messages (the conversation history), model, and prompt (the latest user text). There are no A2A message parts to unpack — the control plane creates a run and your worker claims it, so a chat handler reads the same run.input dict as every other handler.

To stream the answer as it is produced rather than returning it whole, emit it through the ambient client; the control plane relays each new piece of the run's answer text to the browser as it appears.

Event Hooks — rewriting or withholding telemetry before it leaves

Subclass AgentOpsHook and pass instances to the worker. Each callback may return the event unchanged, return a modified one, or return None to drop it — so a tool payload carrying customer data can be rewritten or withheld, and nothing downstream sees the original. before_flush does the same for a whole batch; after_flush reports what was sent.

Every event this worker records crosses one place on its way out, so a hook sees all of it — including the spans the adapters record ambiently, which never pass through your code.

from komodor_agentops import AgentOpsWorker
from komodor_agentops.core.events import AgentOpsEvent
from komodor_agentops.core.hooks import AgentOpsHook


class RedactToolInput(AgentOpsHook):
    async def before_tool(self, event):
        return AgentOpsEvent(**{**vars(event), "payload": {"input": "[redacted]"}})


AgentOpsWorker(agent=spec, on_run=on_run, hooks=[RedactToolInput()]).run()

A hook that raises is logged and its event passes through unchanged, so a buggy rule cannot cost a run its whole transcript. Pass strict_hooks=True to invert that: for a hook whose job is to withhold data, passing the event through on failure ships exactly what it existed to remove, so the failure should propagate instead.

Framework Adapters

LangChain

Spans for each chain, model and tool call, nested under the run's own span. It needs no client: the events go to the ambient buffer the worker's flush cycle already drains, so it only produces spans inside a run scope.

from komodor_agentops.adapters.langchain import ContextCallbackHandler

chain.invoke(input, config={"callbacks": [ContextCallbackHandler()]})

Claude Code

SDK hooks for claude-agent-sdk:

from komodor_agentops.adapters.claude_code.sdk_hooks import agent_ops_hooks_for_run

hooks = agent_ops_hooks_for_run("run_1")

Architecture

komodor-agentops
  agentops-rpc, agentops-otel (wire types + OTel bootstrap)
  httpx, pydantic, pydantic-settings, croniter, pyyaml, python-frontmatter
    |
    +-- [langchain]   -> langchain-core
    +-- [claude-code] -> claude-agent-sdk
    +-- [adk]         -> google-adk
    +-- [agno]        -> agno, anthropic
    +-- [server]      -> no-op alias (kept so existing [server] refs resolve; the SDK
                         serves no HTTP, so there is nothing for it to pull in)

Package Structure

src/komodor_agentops/
  __init__.py       # Public API — the root is the API; submodule paths are internal
  worker_runtime.py # AgentOpsWorker: connect, register, claim, run, report
  worker_client.py  # The worker protocol, and the run scope
  messaging.py      # The transport's composition root (downlink + uplink)
  client.py         # Event buffer + flush
  py.typed          # PEP 561 marker
  transport/        # The byte-level connection (SSE, reconnect)
  bus/              # Frames: the envelope, the router, the uplink
  machinery/        # Delivery: batching, retry, sequencing
  core/             # Context, events, secrets, types, @observe
  worker/           # Run contract, liveness, local runner, log forwarding
  adapters/         # Claude Code, ADK, Agno, LangChain, nanobot

AWS credentials for a worker

An AWS-touching worker needs one variable on the process it runs the AWS CLI or an AWS SDK in:

from komodor_agentops.worker.aws_credentials import serve_credentials

async def fetch():
    # `.model_dump()`: the responder serves AWS's own container-credentials JSON, so it takes that
    # dict rather than the reply model — the field names are already AWS's, so this is a shape
    # change and not a mapping.
    session = await client.aws_credentials()
    return session.model_dump() if session is not None else None

async with serve_credentials(fetch) as creds:
    env = {**scrubbed_env, "AWS_CONTAINER_CREDENTIALS_FULL_URI": creds.uri}

The SDK does the rest — fetch, cache, refresh — because that variable is AWS's own container-credentials contract. Nothing else is needed: no keys in the environment, no per-worker credential code, and no AWS identity on the pod.

serve_credentials asks lazily, on the first fetch rather than on entry, so a worker that never touches AWS costs nothing. client.aws_credentials() returns None when the account has no AWS (IAM role) connection or the worker has no run in flight; that becomes a 503 on the endpoint rather than an exception, because both are ordinary configurations. A retryable failure — a throttled STS — raises instead, and the responder answers 503 while keeping any session it still holds: None means "there is nothing here", a raise means "ask again".

The one thing that bites: the variable starts with AWS_, so any env-blanking helper wipes it. If a worker scrubs ambient AWS variables before launching its subprocess — aws_investigator does, via _scrub_aws_env — set the URI after the scrub, not before.