Authoring a Skill

How to build a Skill plugin against skill_base — the runtime contract for automated workflows.

A Skill is the unit of automation on the platform: it reacts to triggers, reads/writes the entity graph, calls LLM agents, performs actions through connectors, and optionally requests human approval.

Package skeleton

packages/skill_<name>/
├── pyproject.toml
├── src/skill_<name>/
│   ├── __init__.py
│   ├── manifest.yaml
│   └── skill.py
└── tests/
    ├── conftest.py
    └── test_skill.py

pyproject.toml:

[project]
name = "skill_pr_review"
version = "0.1.0"
description = "Review pull requests with LLM agents."
requires-python = ">=3.12"
dependencies = [
    "skill_base",
    "shared_types",
]

[project.entry-points."knowledge_os.skills"]
pr_review = "skill_pr_review:PRReviewSkill"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/skill_pr_review"]

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

The Skill Protocol

skill_base.Skill is structurally typed:

class Skill(Protocol):
    manifest: SkillManifest
    state_kind: str | None
    async def execute(self, ctx: SkillContext) -> SkillResult: ...
    async def on_state_event(
        self, ctx: SkillContext, event: StateEvent,
    ) -> SkillResult: ...

• execute is the entry point fired by every trigger.
• state_kind is None for stateless skills. Set it to a string label (e.g. "conversation") to opt in to persistent state across runs.
• on_state_event is only called for stateful skills. Stateless skills are free to raise NotImplementedError.

The Protocol is not runtime_checkable — mypy + the contract tests enforce the shape; there are no isinstance checks at runtime.

The manifest

A Skill manifest has two top-level keys: plugin: and skill:.

plugin:
  id: pr_review
  type: skill
  version: "0.1.0"
  kernel_version: ">=0.1.0,<2.0"
  display:
    name: PR Review
    description: AI-assisted pull request review.
    category: code_review
    tags: [code, review, ai]
  author:
    name: Threadory

skill:
  triggers:
    - type: entity_event
      entity_kind: pull_request
      events: [created, updated]
    - type: webhook_relay
      connector: my
      events: [merge_request.open, merge_request.update]
    - type: manual

  requires:
    entity_kinds:
      - kind: pull_request
        capabilities: [read_thread, read_diff, read_file_tree]
    actions:
      - kind: pull_request
        capabilities: [post_inline_comment, post_comment]

  optional:
    domain_packs: [pack_python_web]

  config_schema:
    type: object
    properties:
      profile:
        type: string
        enum: [chill, default, assertive, nitpicky]
        default: default
      max_findings:
        type: integer
        default: 5
        minimum: 1
        maximum: 50
    additionalProperties: false

  approval_channels:
    - id: review_post
      title: Post review to PR
      payload_schema_ref: schemas/review_post_v1.json
      default_decision_right: human_required

  exposes:
    sub_skills:
      - id: review_pr
      - id: explain_finding

  cost_hints:
    expected_usd_per_run: 0.10
    typical_duration_sec: 30

  state_kind: null

TriggerSpec discriminated union

A skill must declare at least one trigger. Five concrete types are supported:

type	Fields	Behaviour
entity_event	entity_kind, events	React to graph-level events on an entity kind
webhook_relay	connector, events	Relay a connector webhook into the skill
cron	schedule (5-field cron)	Fire on schedule
manual	—	Only the UI / API can start it
sub_skill	—	Only another skill can start it via ctx.run_skill

The cron expression must have exactly five whitespace-separated fields; pydantic rejects everything else at manifest load.

requires and optional

requires declares the capabilities the skill needs to run. The runtime cross-checks installed connectors before enabling the skill. If no installed connector provides the required capability, the skill is shown as unavailable.

optional.domain_packs lets the skill use a pack's agents when the pack is active in the project, without making it a hard requirement.

config_schema

A JSON Schema draft-2020-12 document. The skill manifest validator runs Draft202012Validator.check_schema on it at load time, so syntactic mistakes fail loudly during import. The UI generates a form from this schema and supplies the validated config to the workflow context.

approval_channels

Each entry binds an id (referenced by ctx.approve(approval_type=...)) to a title and a payload_schema_ref (a path/URI pointing to a JSON Schema for the payload sent to the approver). default_decision_right: auto means the runtime auto-approves unless the project policy says otherwise. human_required forces a human in the loop.

exposes.sub_skills

If your skill wants to be callable as a sub-workflow from other skills, list the sub-skill ids it exposes. The runtime checks the caller's permissions against this list.

cost_hints

Soft hints surfaced in the UI for budgeting. expected_usd_per_run and typical_duration_sec must be non-negative.

state_kind

null for stateless. Set to a non-empty string when the skill is conversational and needs ctx.state to persist across runs.

The @skill decorator

The @skill decorator loads manifest.yaml at import time, attaches it to the class as cls.manifest, and validates that manifest.plugin.id matches the name= you pass in:

from skill_base import skill


@skill(manifest_path="manifest.yaml", name="pr_review")
class PRReviewSkill:
    state_kind: str | None = None

    async def execute(self, ctx):
        ...

The manifest_path is resolved relative to the source file of the decorated class. Ship manifest.yaml next to skill.py and you're done — no pkg_resources calls in your code.

If the name= you pass does not match manifest.plugin.id, the decorator raises ValueError immediately at import time. This catches renames before they reach production.

The SkillContext — 10 sub-APIs

SkillContext is the everything-bag the runtime passes to execute. The Protocol lives in skill_base.context. Top-level attributes:

class SkillContext(Protocol):
    project: Project
    organization: Organization
    trigger: TriggerEvent
    entity: Entity | None
    domain_pack: Any | None

    entities: EntitiesAPI
    retrieval: RetrievalAPI
    agents: AgentRunner
    actions: ActionAPI
    connectors: ConnectorRegistry
    approve: ApprovalRequester
    run_skill: SubSkillRunner
    state: SkillState
    telemetry: TelemetryAPI
    logger: Logger

Sub-API	Use it for
entities	CRUD on the entity graph (create, upsert, get, query, link, bulk_upsert)
retrieval	Semantic + structural retrieval (query, deterministic)
agents	Run a named LLM agent: await ctx.agents.run(agent="walkthrough", input=...)
actions	Execute capability-typed writes through the policy engine
connectors	Low-level connector handles by id or capability (escape hatch)
approve	Suspend the workflow until a human approves
run_skill	Start another skill as a Temporal sub-workflow
state	Per-run persistent key/value store (only meaningful when state_kind set)
telemetry	OTel spans + custom metrics
logger	Structured logger pre-bound with project / workflow context

Skills must touch the platform only through this surface. Direct imports from kernel.*, sqlalchemy, or temporalio inside skill code are bugs.

Stateless skill — a worked example

from __future__ import annotations

from shared_types import SkillResult
from skill_base import skill
from skill_base.context import SkillContext


@skill(manifest_path="manifest.yaml", name="meeting_notes")
class MeetingNotesSkill:
    state_kind: str | None = None

    async def execute(self, ctx: SkillContext) -> SkillResult:
        meeting = ctx.entity
        if meeting is None:
            return SkillResult(status="failed", summary="No meeting entity in trigger")

        if (meeting.metadata or {}).get("duration_sec", 0) < 300:
            return SkillResult(
                status="completed",
                summary="Meeting too short, skipped",
                metrics={"skipped": 1.0},
            )

        walkthrough = await ctx.agents.run(
            agent="walkthrough",
            input={"transcript": meeting.body or ""},
        )

        decisions = await ctx.agents.run(
            agent="decision_extractor",
            input={"transcript": meeting.body or "", "walkthrough": walkthrough.output},
        )

        for item in decisions.output.get("items", []):
            art = await ctx.entities.create(
                kind="decision",
                source_system="internal",
                source_id=f"{meeting.id}/decision/{item['local_id']}",
                title=item["title"],
                body=item.get("reasoning"),
                metadata={"decision": item["decision"], "owner": item.get("owner")},
            )
            await ctx.entities.link(
                source=meeting.ref,
                target=art.ref,
                relation="discusses",
                confidence="high",
                discovery_source="ai_extraction",
            )

        return SkillResult(
            status="completed",
            summary=f"Extracted {len(decisions.output.get('items', []))} decisions",
            metrics={"decisions": float(len(decisions.output.get("items", [])))},
        )

Stateful skill — opt in to state_kind

Set state_kind in the manifest and on the class. The runtime then guarantees:

• ctx.state.get/set/delete/all survive across execute and on_state_event invocations.
• on_state_event is delivered when an external event for this state instance arrives.

skill:
  triggers:
    - type: manual
  state_kind: conversation

from skill_base import skill
from skill_base.protocol import StateEvent


@skill(manifest_path="manifest.yaml", name="support_thread")
class SupportThreadSkill:
    state_kind: str | None = "conversation"

    async def execute(self, ctx):
        await ctx.state.set("turn", 0)
        await ctx.state.set("history", [])
        return SkillResult(status="awaiting_approval", summary="Listening")

    async def on_state_event(self, ctx, event: StateEvent):
        history = list(await ctx.state.get("history") or [])
        history.append({"type": event.type, "payload": event.payload})
        await ctx.state.set("history", history)
        # ... reply, update entities, etc.
        return SkillResult(status="completed", summary="Replied")

Composing skills with ctx.run_skill

Sub-skills run as Temporal sub-workflows. Wait for completion (default) or fire-and-forget:

result = await ctx.run_skill(
    name="dejavu_detector",
    input={"text": ticket.body or ""},
)
# `wait="started"` returns immediately once Temporal schedules the child.

The called skill must declare a sub_skill trigger and be listed in the caller's permitted exposes.sub_skills set on the project.

Requesting approval

approval = await ctx.approve(
    approval_type="review_post",                 # must match manifest approval_channels[].id
    title="Post AI review to PR #42",
    payload={"findings": [...]},
    payload_schema_ref="schemas/review_post_v1.json",
    channel="internal",
    expires_in_seconds=3600,
)
# When this returns, the approval is "decided" (approved, rejected, or expired).

If the policy engine auto-approves, the call returns near-instantly. Otherwise the workflow pauses inside Temporal until a human resolves it.

Telemetry and logging

Use spans around expensive blocks:

async with ctx.telemetry.span("classify_finding", attributes={"profile": "default"}) as span:
    result = await ctx.agents.run(agent="classifier", input=...)
    span.set_attribute("agent.cost_usd", result.cost_usd)

ctx.logger is a stdlib Logger already bound with structured fields the runtime supplies (project_id, workflow_run_id). Just call ctx.logger.info("...").

Output contract

Always return a SkillResult:

class SkillResult(BaseModel):
    status: Literal["completed", "failed", "awaiting_approval", "cancelled"]
    summary: str
    artifacts: list[ArtifactRef] = []
    metrics: dict[str, float] = {}
    next_runs: list[ScheduledRun] = []

Pick the status that matches what actually happened — the UI shows the summary in the activity feed, and metrics is forwarded to telemetry.

Testing skills

testing_kit.KernelMock is an in-memory implementation of every sub-API. FakeLLMRouter records agent calls and returns canned outputs. See Testing for the full pattern.

Smoke test:

from uuid import uuid4

import pytest
from testing_kit import KernelMock, FakeLLMRouter

from skill_my import MySkill


async def test_executes_happy_path() -> None:
    mock = KernelMock(project_id=uuid4())
    router = FakeLLMRouter()
    router.register_fixture("walkthrough", {"summary": "..."})
    router.register_fixture("decision_extractor", {"items": [...]})
    mock._fake_llm_router = router  # type: ignore[attr-defined]

    skill = MySkill()
    result = await skill.execute(mock)

    assert result.status == "completed"
    router.assert_called_with("walkthrough")
    assert any(call["action"].kind == "..." for call in mock.actions_called) is False

The reusable contract test

Inherit from SkillContractTest (re-exported from both skill_base and testing_kit) to gain a standard assertion suite that mirrors what the runtime checks at registration time:

from skill_base import SkillContractTest

from skill_my import MySkill


class TestMySkillContract(SkillContractTest):
    __test__ = True  # opt back in — base class sets it False
    skill_cls = MySkill

The base class verifies:

• manifest is an instance of SkillManifest (i.e. @skill was applied).
• state_kind is None or a str.
• execute(ctx) exists, is a coroutine function, and takes exactly one arg called ctx.
• execute is annotated to return SkillResult.
• If state_kind is set, on_state_event exists and is a coroutine function.

Skill checklist

• [ ] Class is decorated with @skill(manifest_path=..., name=...).
• [ ] state_kind matches between manifest and class.
• [ ] At least one TriggerSpec declared.
• [ ] All required entity kinds + actions listed in requires are reachable through ctx.
• [ ] No imports from kernel.*, sqlalchemy, temporalio, or any connector implementation.
• [ ] execute always returns a SkillResult — including in failure paths.
• [ ] Approvals use ids declared in approval_channels.
• [ ] Sub-skill IDs in exposes match what other skills call via ctx.run_skill.
• [ ] SkillContractTest subclass passes.
• [ ] Entry point registered under knowledge_os.skills.