Python SDK - Komodor

The Python SDK is how you turn your own code into an agent. It provides the worker runtime, the agent definition, and helpers for producing well-formed output and evidence. This page walks the path: install, define, implement, emit evidence, test locally, deploy.

Install

Agents are built with the komodor-agentops package, on Python 3.11 or newer. The SDK is framework-agnostic — a plain function, a chain, a tool-using loop — and adapters ship as extras:

pip install komodor-agentops                # core + worker runtime
pip install "komodor-agentops[langchain]"   # LangChain adapter
pip install "komodor-agentops[claude-code]" # Claude Agent SDK adapter
pip install "komodor-agentops[adk]"         # Google ADK adapter
pip install "komodor-agentops[agno]"        # Agno adapter
pip install "komodor-agentops[all]"         # everything

Project layout

The agent’s identity and knowledge live as files next to the code:

my_agent/
  agent-spec.yaml     # identity, labels, input schema, triggers (required)
  agent.md            # optional agent context
  worker.py           # the handler + entrypoint
  skills/
    triage.md         # optional skills, published to the catalog
    rca/SKILL.md

agent-spec.yaml is the source of truth for identity:

schema_version: 1
agent_id: incident-summarizer
name: Incident Summarizer
description: Summarizes incidents into severity, root cause, and next steps.
owner: my-team
labels:
  category: incident-response
default_input:
  prompt: Summarize the latest incident.
triggers:
  - id: nightly
    type: schedule
    name: Nightly summary
    cron: "0 6 * * *"

AgentSpec.from_dir(...) loads the spec, then picks up agent.md and skills/ automatically. Declared triggers and skills are registered when the worker connects.

A minimal worker

A worker is three things: an agent spec, an async handler turning a run’s input into its output, and a run call.

from pathlib import Path
from typing import Any

from komodor_agentops import AgentOpsWorker, AgentSpec
from komodor_agentops.worker import Run

AGENT_SPEC = AgentSpec.from_dir(Path(__file__).parent)

async def handler(run: Run) -> dict[str, Any]:
    prompt = run.input.get("prompt", "")

    summary = await summarize_incident(prompt)

    return {
        "severity": summary.severity,
        "root_cause": summary.root_cause,
        "text": summary.markdown,
    }

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

if __name__ == "__main__":
    main()

.run() configures logging, pulls the agent’s credentials at boot, dials the control plane, and begins heartbeating. To embed the worker in an existing application, await AgentOpsWorker(...).serve() instead.

The worker binds no port and serves nothing. It dials out and receives its runs over that one connection — no HTTP server to expose, no inbound rule, nothing for an ingress to point at.

Raising from the handler fails the run, with the exception as the reported error. There is no “return a failed result” form. summarize_incident is your own application function and must be implemented or imported — this is an integration skeleton, not a complete incident-analysis implementation. A genuinely minimal handler can just return what it was given.

The output contract

A result can carry structured fields and descriptive text, and getting this right is what makes an agent usable in more than one place:

Channel	What it is	Who reads it
Structured	Your own top-level keys in the returned dict	A script, an MCP client, or a downstream workflow step making a decision
Descriptive	The reserved `text` key, holding markdown	A person, in the console, in Slack, in history

Both live in the one dict the handler returns. An automation-only agent can return structured fields alone; an agent whose result a human reads should provide both. Keep them aligned — do not bury machine values inside prose, and do not invent a second human-text field, because text is reserved. When an agent provides no descriptive text, the platform renders the structured output readably instead.

Emitting evidence

While it runs, your agent can stream the same evidence the built-in agents do:

Spans — wrap a function with @observe(name="...", as_type="tool") to record nested trace spans.
Streaming text — await stream_status_text("...") streams partial output to streaming-capable callers such as chat; it is a no-op elsewhere.
Artifacts — attach files and reports to the run through the artifact client.

With a framework adapter most of this is automatic: the adapters record model calls, tool calls, and token usage without extra code.

Credentials

Do not bake keys into the image. Store them as credentials, bind them to the agent, and the SDK delivers them:

Per run — bound credentials arrive with each run; call apply_secret_to_env("ANTHROPIC_API_KEY") or get_secret(...) in your handler.
At boot — for a token needed before any run, nothing is required: the worker pulls the on-demand credentials bound to it at start-up.

Values delivered this way are masked out of the evidence the worker emits; see Secrets & credential handling.

Test locally

Run your handler once, with no control plane, using the CLI installed with the SDK:

agentops-run-worker \
  --handler my_agent.worker:handler \
  --input-json '{"prompt": "Summarize incident 42"}' \
  --env-file .env \
  --output output.json --artifact-dir artifacts/

It loads env files, exposes detected secrets as run-scoped secrets and puts them in the environment the way a real run does, calls your handler with a Run built from the flags, and writes the output. Anything uploaded with run.upload_artifact(...) lands in --artifact-dir. The same flow is available programmatically as run_worker_once and run_worker_once_sync.

Deploy

A worker is a normal service you deploy however you run software. It needs two things from the platform, both as environment variables:

Variable	Meaning
`AGENTOPS_URL`	The control-plane base URL
`AGENTOPS_WORKER_TOKEN`	A worker token minted for this agent

Use cases

An agent can advertise named, schema-typed actions in its agent-spec.yaml, and each becomes a callable tool on that agent’s own MCP endpoint:

use_cases:
  - slug: triage_build
    description: Triage a failed CI build and summarize the likely cause.
    input_schema:
      type: object
      properties:
        build_id: { type: string }
      required: [build_id]
    timeout_seconds: 300

Read them back off a stored agent card with UseCase.list_from_card(agent_card). Every agent also advertises two built-ins — a connectivity check and a sample run — without declaring anything. See Use cases for the full field list and how they are called.

Next steps

Go SDK

The same worker contract, without Python in the container.

Agent identity

Where the worker token comes from, and how to rotate it.

Runs & evidence

What the evidence you emit becomes.

Skills

The skills/ directory, and what happens to it.

​Install

​Project layout

​A minimal worker

​The output contract

​Emitting evidence

​Credentials

​Test locally

​Deploy

​Use cases

​Next steps

Go SDK

Agent identity

Runs & evidence

Skills

Install

Project layout

A minimal worker

The output contract

Emitting evidence

Credentials

Test locally

Deploy

Use cases

Next steps