tmforum-oda/oda-canvas/create-oda-operator

Name: create-oda-operator
Author: tmforum-oda

Claude Code and agent skill by tmforum-oda/oda-canvas

Trust tier 3

community

Source check pending

Checking that the SKILL.md file is still in the repository.

Guide for creating Kubernetes operators for the ODA Canvas using the Python KOPF framework. Use this skill whenever you are building or modifying any Canvas operator, adding a kopf handler to existing code, creating a CRD watcher, implementing component lifecycle management, writing any Python Kubernetes controller, working with files in source/operators/, or answering questions like "how do I watch a CRD", "how do I update status", "how should I structure my operator chart", or "how do I add a health check to my operator". Also use for questions about operator peering, admission webhooks, timer/periodic handlers, or any kopf framework question.

Install

npx vskill@latest install tmforum-oda/oda-canvas/create-oda-operator

SKILL.md

Create ODA Operator

Technology

Most ODA Canvas operators use KOPF — Kubernetes Operators Framework for Python.

The Go-based operators (pdb-management/, oauth2EnvoyfilterOperator/) are outside this skill's scope.

Core dependencies (pinned versions used by this project):

kopf==1.37.2
kubernetes==27.2.0
python-json-logger==2.0.7
PyYAML
requests              # for external HTTP calls

Additional per-operator: cloudevents, jinja2, hvac (HashiCorp Vault), google-auth.

References:

`references/kopf-docs.md` — curated kopf documentation links by topic
`references/operator-catalog.md` — all existing operators with file paths and patterns

Quick Reference: Existing Operators

Before writing new code, check if an existing operator matches your use case:

Use case	Reference operator
Component/CRD lifecycle, sub-resource management	`source/operators/TMFOP001-Component-Management/component-management/`
API gateway integration (HTTPRoute, Kong plugins)	`source/operators/TMFOP002-API-Management/kong/`
Service mesh integration (VirtualService, EndpointSlice)	`source/operators/TMFOP002-API-Management/istio/`
External identity service integration + health timers	`source/operators/TMFOP003-Identity-Config/keycloak/`
Pod sidecar injection, admission webhooks	`source/operators/TMFOP007-Secrets-Management/vault/`
Cross-resource URL discovery, dependency resolution	`source/operators/TMFOP005-Dependency-Management/simple-dependency-management/`
Field-triggered handler, Kubernetes Secret creation	`source/operators/TMFOP004-Credentials-Management/credentials-management/`

See references/operator-catalog.md for kopf decorators used and detailed patterns in each.

Directory Structure

Operators are organised under source/operators/ in two-level directories.

Modern Pattern (Recommended)

For new operators, use a docker/ subdirectory with standard Dockerfile naming:

source/operators/TMFOP###-<Domain>/
  <implementation>/
    README.md                   # Operator documentation
    docker/                     # Docker build context
      Dockerfile                # Standard Docker filename (capital D)
      requirements.txt          # Python dependencies
      src/                      # Python source files
        <operatorName>.py       # Main operator file
        log_wrapper.py          # Structured logging helper
        ...                     # Other Python modules
      templates/                # Optional Jinja2 templates
      DockerfileTest            # Optional test Dockerfile

Example: source/operators/TMFOP005-Dependency-Management/simple-dependency-management/

Advantages:

Follows standard Docker conventions (Dockerfile with capital D)
Clean separation of build context
Supports multi-stage builds and test dockerfiles
No need for explicit buildDockerfile in workflow config

Legacy Pattern (Existing Operators)

Older operators use files in the operator root directory:

source/operators/TMFOP###-<Domain>/
  <implementation>/
    <operatorName>.py           # Main operator file
    log_wrapper.py              # Structured logging shared helper
    <operatorName>-dockerfile   # Dockerfile (no extension, lowercase)
    requirements.txt            # Optional per-operator deps

Examples: component-management/, kong/, keycloak/

Note: When maintaining legacy operators, keep the existing pattern for consistency. New operators should use the modern pattern.

The matching Helm chart lives at charts/<operator-chart-name>/.

Constants & Startup

Define CRD coordinates and HTTP codes at the top of every operator:

python

GROUP = "oda.tmforum.org"
VERSION = "v1"
COMPONENTS_PLURAL = "components"   # or "exposedapis", "dependentapis", "identityconfigs", etc.
HTTP_CONFLICT = 409
HTTP_NOT_FOUND = 404

Always configure the watcher timeout in a startup hook to recover from stale watch connections:

python

@kopf.on.startup()
def configure(settings: kopf.OperatorSettings, **_):
    settings.watching.server_timeout = 1 * 60  # recover from broken watchers every 60s
    # Optional — see https://docs.kopf.dev/en/stable/configuration/
    # settings.posting.level = logging.WARNING    # reduce K8s event verbosity
    # settings.peering.priority = 100             # set if running alongside other operators

Handler Patterns

Create / Resume / Update (triple-stacked)

The standard CRD lifecycle pattern — one function handles all three events. resume fires when the operator starts and finds pre-existing objects, ensuring continuity after restarts.

python

@kopf.on.resume(GROUP, VERSION, COMPONENTS_PLURAL, retries=5)
@kopf.on.create(GROUP, VERSION, COMPONENTS_PLURAL, retries=5)
@kopf.on.update(GROUP, VERSION, COMPONENTS_PLURAL, retries=5)
async def coreAPIs(meta, spec, status, body, namespace, labels, name, **kwargs):
    pass

Docs: https://docs.kopf.dev/en/stable/handlers/

Delete Handlers

python

@kopf.on.delete(GROUP, VERSION, COMPONENTS_PLURAL, retries=5)
async def componentDelete(meta, spec, status, namespace, name, **kwargs):
    pass

Kopf adds a finalizer so the object isn't fully deleted until this handler succeeds. Pass optional=True if the handler shouldn't block object deletion.

Field Handlers

React to specific field changes. Use a when= callback to conditionally filter to avoid re-processing unchanged states:

python

def is_status_changed(status, **kwargs):
    return status.get("identityConfig") != "In-Progress-Done"
@kopf.on.field(GROUP, VERSION, COMPONENTS_PLURAL,
               field="status.identityConfig",
               when=is_status_changed, retries=5)
async def identityConfig(meta, spec, status, old, new, namespace, **kwargs):
    pass

Docs: https://docs.kopf.dev/en/stable/filters/

Timer Handlers

Run periodically as long as the object exists — useful for health polling and reconciliation. idle waits for the resource to stabilise before first invocation:

python

@kopf.timer(GROUP, VERSION, "identityconfigs", interval=300.0, idle=30.0)
async def healthProbe(spec, status, namespace, name, **kwargs):
    # Periodic reconciliation / external service health check
    pass

Docs: https://docs.kopf.dev/en/stable/timers/

Health Check Probes

Expose a health endpoint Kubernetes liveness probes can call:

python

@kopf.on.probe(id="http-check")
def check_http_connection(**kwargs) -> str:
    resp = requests.get(SERVICE_URL, timeout=10)
    resp.raise_for_status()
    return "ok"

Docs: https://docs.kopf.dev/en/stable/probing/

Low-Level Event Watching

Watch raw watch-stream events without kopf storing handler state — useful for mirroring changes from another operator's CRDs or for lightweight reactive logic:

python

@kopf.on.event(GROUP, VERSION, "exposedapis")
def onExposedAPIEvent(event, body, **kwargs):
    if event["type"] in (None, "ADDED", "MODIFIED"):
        pass  # react to changes; no retry/status tracking by kopf

Docs: https://docs.kopf.dev/en/stable/handlers/#event-watching-handlers

Admission / Mutation Webhooks

Intercept resource creation/updates before they reach the cluster — the secretsmanagement operator uses this to inject a Vault sidecar into pods:

python

@kopf.on.mutate("", "v1", "pods", operation="CREATE")
async def inject_sidecar(patch, spec, namespace, **kwargs):
    if should_inject(spec):
        patch.spec.setdefault("containers", []).append(SIDECAR_CONTAINER)

Requires a webhook server configured in startup. In-cluster, use a ServiceTunnel so Kubernetes calls back via the operator's Service DNS name. For simpler cases kopf can manage the webhook configuration automatically:

python

@kopf.on.startup()
async def configure(settings: kopf.OperatorSettings, **_):
    settings.admission.server = kopf.WebhookServer(addr="0.0.0.0", port=8080)
    settings.admission.managed = "auto.kopf.dev"
    settings.peering.priority = 200  # ensure admission operator has highest priority

See source/operators/TMFOP007-Secrets-Management/vault/docker/ for the full ServiceTunnel implementation. Docs: https://docs.kopf.dev/en/stable/admission/

Sub-Handlers

Process a list of items with per-item kopf lifecycle (retries, status, logging):

python

@kopf.on.create(GROUP, VERSION, COMPONENTS_PLURAL)
async def processAPIs(spec, **_):
    for api in spec.get("coreFunction", {}).get("exposedAPIs", []):
        @kopf.subhandler(id=api["name"])
        def handle_api(api=api, **kwargs):
            pass

Docs: https://docs.kopf.dev/en/stable/handlers/#sub-handlers

Error Handling

Three error categories — pick the right one to control retry behaviour:

python

# Transient failure — kopf will retry after delay (default delay is 60s)
raise kopf.TemporaryError("External service not ready", delay=30)
# Permanent failure — stop retrying, mark handler as failed
raise kopf.PermanentError("Invalid spec: field X is required")
# Configure error mode per handler (default is TEMPORARY = retry forever)
@kopf.on.create(GROUP, VERSION, PLURAL,
                errors=kopf.ErrorsMode.PERMANENT,
                retries=5,
                backoff=30)
async def handler(**_): ...

Always handle ApiException explicitly when calling the Kubernetes client:

python

from kubernetes.client.rest import ApiException
try:
    resource = custom_api.get_namespaced_custom_object(...)
except ApiException as e:
    if e.status == HTTP_NOT_FOUND:
        pass  # create it
    else:
        raise kopf.TemporaryError(str(e), delay=30)

Docs: https://docs.kopf.dev/en/stable/errors/

Status Updates & Results Delivery

A handler's return value is automatically written to status.<handler_id>. Use a dict to store structured results:

python

@kopf.on.create(GROUP, VERSION, COMPONENTS_PLURAL, retries=5)
async def coreAPIs(name, **kwargs):
    result = await processAPIs(...)
    return {"apiCount": len(result), "status": "Active"}  # → status.coreAPIs = {...}

Docs: https://docs.kopf.dev/en/stable/results/

Kubernetes API Calls

python

import kubernetes
# Custom resources (CRDs)
coa = kubernetes.client.CustomObjectsApi()
# Idempotent create-or-update pattern
try:
    coa.get_namespaced_custom_object(GROUP, VERSION, namespace, PLURAL, name)
    coa.patch_namespaced_custom_object(GROUP, VERSION, namespace, PLURAL, name, body)
except ApiException as e:
    if e.status == HTTP_NOT_FOUND:
        coa.create_namespaced_custom_object(GROUP, VERSION, namespace, PLURAL, body)
    else:
        raise kopf.TemporaryError(str(e), delay=30)
# Core Kubernetes resources
core_v1 = kubernetes.client.CoreV1Api()
core_v1.create_namespaced_secret(namespace, secret_body)
# List cluster-wide resources for cross-resource discovery
all_apis = coa.list_cluster_custom_object(GROUP, VERSION, "exposedapis")
# Watch non-CRD resources (e.g. EndpointSlice for service implementation discovery)
@kopf.on.create("discovery.k8s.io", "v1", "endpointslices", retries=5)
@kopf.on.update("discovery.k8s.io", "v1", "endpointslices", retries=5)
async def onEndpointSlice(meta, spec, namespace, name, **kwargs):
    pass

Docs: https://docs.kopf.dev/en/stable/idempotence/

Resource Hierarchies

Use kopf.adopt() to set ownerReferences on child resources so they are deleted with the parent (cascaded deletion):

python

child = {"apiVersion": "oda.tmforum.org/v1", "kind": "ExposedAPI", ...}
kopf.adopt(child)   # sets ownerReferences, namespace, and labels from current handler context
coa.create_namespaced_custom_object(GROUP, VERSION, namespace, "exposedapis", child)

Individual hierarchy helpers when you need fine-grained control:

This is the start of the file. Read the full SKILL.md on GitHub.

Security

Security scanPassed, 95 of 100Pattern checks on the skill's files.
Model reviewPassed

See the full security report

Quality evaluation

Task evaluationNot runNo evaluation yet. Run one in Skill Studio.

See evaluation results

Source

SKILL.md in the repositoryCheckingskills/create-oda-operator/SKILL.md

Popularity

Repository stars: 50
Repository forks: 67
7-day trend score: 20

Stars and forks are for the whole GitHub repository, not this skill alone. Refreshed 25 May 2026, 5 months ago.

Works with

Any agent vskill installs to: 56 targets in all. These 39 have a page on this site.

Claude Code
Cursor
GitHub Copilot
Windsurf
Codex
Gemini CLI
Cline
Amp
Roo Code
Goose
Continue
Kilo Code
Junie
Trae
OpenHands
Qwen Code
AI Agent
Tabnine Chat
Sourcegraph Cody
Amazon Q
Firebase Genkit
Sweep
Devin
Phind
Mentat
Claude Desktop
Aider
GPT Pilot
Plandex
bolt.new
v0
Replit Agent
Anse
TypingMind
BetterChatGPT
LibreChat
Warp AI
Pieces
Fig AI

Embed badge

[![vskill VERIFIED](https://verified-skill.com/api/v1/skills/tmforum-oda/oda-canvas/create-oda-operator/badge)](https://verified-skill.com/skills/tmforum-oda/oda-canvas/create-oda-operator)

Something wrong with this skill or its page? Tell us.