Skip to content

Examples

The repository ships a set of runnable specs and supporting Python helpers under examples/. Use them to validate a local install, exercise the CLI, or as starting points for your own specs.

These specs are non-production placeholders — the provider/model fields point at common defaults and should be retargeted before any real use.

Run the commands below from the repository root. A spec resolves its tools by a dotted implementation: path (e.g. examples.app_functions.web_search), which is imported relative to your current working directory. The CLI puts the invocation directory on the import path for you, so kneo spec compile examples/research_agent.yaml works as written from the repo root — and a spec of your own resolves its implementation: modules relative to your project root the same way.

Feature → example matrix

Pick an example by the feature or surface you want to see. The regression examples double as the executable proof of a specific fix and run offline in CI (no provider keys, no network).

Feature / surface Example What it shows
Single agent + tools, env overlays research_agent.yaml (+ .dev/.staging/.prod) A base agent with function tools and per-environment overlays.
Graph workflow graph_review_workflow.yaml Conditional edges between steps.
Concurrent fan-out concurrent_review_workflow.yaml Parallel branches that join.
Group chat group_chat_workflow.yaml Multi-agent turn-taking.
Human-in-the-loop human_approval_workflow.yaml Pause at a human step, resume via the continuation API.
Human pause/continuation per workflow shape (0.12.0 § D) graph_human_approval_workflow.yaml, concurrent_human_approval_workflow.yaml, group_chat_human_approval_workflow.yaml, handoff_human_approval_workflow.yaml A human-approval gate pauses + resumes in every workflow shape (graph / concurrent / group-chat / handoff; concurrent via drain-then-block).
Human (smoke) smoke_human_workflow.yaml Minimal pause/resume on the dummy provider, for smoke tests.
Declarative MCP / agent-as-tool / workflow-as-agent declarative_spec.yaml The 0.8.0 declarative-parity features (compiles offline).
Run-level timeouts run_with_timeout.py Deadlines + the prune_timed_out_runs sweep.
Project config / overlays project_config.yaml Per-environment overlays + policy + retention knobs.
Handoff round_robin (0.9.0 regression) handoff_workflow.yaml One turn per participant, then a clean completed.
Per-step on_error (0.9.0 regression) resilient_workflow.yaml fallback and continue policies actually executing.
MCP stdio transport (0.9.0 regression) mcp_stdio_workflow.yaml + mcp_stdio_server.py A real subprocess + stdio handshake on first tool call.
Nested workflow + human approval (0.10.0 HIGH #1) nested_workflow_human_approval.yaml A nested pipeline gated by a top-level human step; the human-inside-nested anti-pattern is rejected at validation.
Guardrail enforcement (0.10.0 HIGH #2; completed 0.11.0) guardrails_complete.yaml A declared tool-stage guardrail redacts PII; as of 0.11.0 a workflow-stage one is enforced per step (no longer rejected).
Secret redaction (0.10.0 MEDIUMs) redaction_demo.py Redaction across structured data, free text, and traces; pluralized credential keys redacted, usage keys preserved.
Custom middleware + adapter hop (0.10.0 MEDIUMs) custom_middleware_demo.py A custom middleware's ToolResult.metadata reaches the SDK; OTel context survives the worker-thread hop.
Spec-path confinement (1.0.0 default-on) confinement_demo.py Caller-supplied spec_path / overlays / skills[].source outside the spec root (KNEO_SERV_SPEC_ROOT, or the working dir) are rejected 422 spec_path_confined.
MCP HTTP/SSE transport mcp_http_workflow.yaml The http (and sse) MCP transport shape, compile-only — complements the runnable stdio proof.
Guardrail input/output stages guardrail_stages.yaml The two request/response-boundary stages: input fail-closed block + output PII redaction (companion to guardrails_complete.yaml).
Guardrail workflow stage (0.11.0) guardrail_workflow_stage.yaml The fourth stage — a per-step workflow-stage guardrail enforced on each step's output (was rejected at validate before 0.11.0).
Human request taxonomy human_task_taxonomy.yaml request_type (review / correction / selection / freeform), options/default_option, context, and timeout_seconds + on_timeout policy on a kind: human step.

Spec files

research_agent.yaml

A single-agent research pipeline using a plan-act strategy with two tools (web_search, webpage_reader) and a sequential workflow that retrieves, analyzes, and summarizes.

kneo spec validate examples/research_agent.yaml
kneo spec compile examples/research_agent.yaml
kneo run --input "Analyze Nvidia AI business" --target workflow examples/research_agent.yaml

Three environment overlays show the overlay system in action:

# `bundle sign` requires the signing key in the environment.
export KNEO_SERV_SPEC_SIGNING_KEY=…
kneo spec validate examples/research_agent.yaml --env prod
kneo spec bundle sign examples/research_agent.yaml \
  --output bundles/research_agent.prod.json --approved-by release-manager --env prod

graph_review_workflow.yaml

A graph workflow with conditional edges: retrieve → analyze → review → revise → finalize, where the review step routes to revise or finalize based on output. Demonstrates GraphWorkflow, conditional edges, and component agent references.

kneo spec compile examples/graph_review_workflow.yaml

concurrent_review_workflow.yaml

A concurrent workflow that fans out a single input to three reviewers (security, accessibility, performance) running in parallel. The platform collects each participant's response and returns the combined result. Demonstrates ConcurrentWorkflow, participants:-style declaration, and the fan-out / fan-in pattern.

kneo spec compile examples/concurrent_review_workflow.yaml
kneo run --input "Review the auth middleware refactor" \
  --target workflow examples/concurrent_review_workflow.yaml

group_chat_workflow.yaml

A group-chat workflow with three personas (proponent, skeptic, pragmatist) debating a design proposal over two rounds. Each round visits all participants in declaration order, so rounds: 2 produces six total turns. Demonstrates GroupChatWorkflow, the rounds: knob, and ordered participant declaration for structured back-and-forth.

kneo spec compile examples/group_chat_workflow.yaml
kneo run --input "Should we adopt gRPC for service-to-service calls?" \
  --target workflow examples/group_chat_workflow.yaml

human_approval_workflow.yaml

Sequential workflow with a human-in-the-loop step (kind: human) between draft and publish. Use it to exercise the pause/resume API.

kneo run --input "hello" --target workflow --json examples/human_approval_workflow.yaml
# Output includes a continuation_id; resume with:
kneo human resume <continuation_id> --request-id <request_id> --approve

The deeper human-task documentation is in design.md § 8.5 and the HTTP API's /human-tasks/... endpoints.

Timeout branches

The approval-reviewer block in this spec declares a 24-hour timeout with on_timeout: escalate. Two other literals are available; the platform dispatches per human_in_the_loop.md § 9:

on_timeout Lifecycle Audit event(s) Continuation
fail (default) Run transitions to expired human.expired deleted
continue Synthesizes an auto-approved HumanResponse and resumes the workflow human.continued; human.continue_failed on resume error deleted on success or failure
escalate Run stays blocked; escalated_at stamped on the continuation; subsequent prune calls skip it (escalation fires once) human.escalated preserved (operator reassigns + resumes via the normal /human-tasks/{continuation_id}/resume path)

Auto-routing of an escalated task to a different reviewer is up to the operator's external workflow — the platform marks + audits the task as escalated, it does not auto-reassign. Operators call PlatformManager.prune_expired_human_tasks() (cron, scheduled run, manual sweep — same pattern as prune_retention(); there is no built-in scheduler) to dispatch the timeout branch.

run_with_timeout.py

Worked walkthrough of the run-level timeout — distinct from the human-task timeout above. start_run_from_spec(..., timeout_seconds=N) schedules a run with a wall-clock deadline written to RunState.deadline_at; prune_timed_out_runs() is the operator-callable sweep that force-cancels every running or blocked run past its deadline, transitions the state to timed_out, deletes any associated continuation, and emits a run.timed_out audit event.

python examples/run_with_timeout.py

Whichever timeout fires first wins. The dispatch matrix between run-level and human-task deadlines is documented in human_in_the_loop.md § 9 under Run-level timeouts vs. human-task timeouts.

smoke_human_workflow.yaml

Lightweight human-in-the-loop spec that uses the dummy provider so it runs without real provider credentials. Used by the deployment smoke script:

python scripts/deployment_smoke.py --base-url http://127.0.0.1:8000

See deployment_smoke.md for the full smoke sequence.

declarative_spec.yaml

The declarative-spec-parity features added in 0.8.0, in one spec:

  • MCP transport — mcp_servers declares an http (or stdio / sse) server, and a tool binds to it with a tool.mcp block. The server config is built at compile time but only connected on first tool call, so it compiles offline. Enterprise mTLS fields (verify / ca_bundle / client_cert / client_key_ref) ride the same block.
  • Agent-as-tool — tool.agent backs a tool with another component agent.
  • Workflow-as-agent — agent.as_agent backs an agent with a workflow.

The build-order graph wires these cross-component references; a forward or cyclic reference is rejected at validate, not at runtime.

kneo spec validate examples/declarative_spec.yaml
kneo spec compile examples/declarative_spec.yaml

Skills have two surfaces. Declared skills are a spec field: a top-level skills: block maps a name to a SkillSpec with a source (the bundle's filesystem path), and an agent references them by name in its skills: list. A declared skills[].source is a caller-supplied path, so it is confined to the spec root (KNEO_SERV_SPEC_ROOT, or the working directory by default) like spec_path/overlays — an out-of-root source is rejected 422 spec_path_confined.

A runnable example: examples/skills_spec.yaml declares a code_review skill sourced from the bundle at examples/skills/code_review/ and activates it on the agent. Compile it from the repo root so the relative source resolves inside the spec root:

kneo spec validate examples/skills_spec.yaml
kneo spec compile examples/skills_spec.yaml

Separately, the runtime/API surface lets you list the discoverable skills and toggle them per run against a running service:

# Read-only catalog (auth scope: specs:read)
curl -H "Authorization: Bearer $KEY" http://127.0.0.1:8000/v1/skills

# Per-request overlay on a run — add/disable within your scope; audited.
curl -X POST http://127.0.0.1:8000/v1/runs \
  -H "Authorization: Bearer $KEY" -H "Content-Type: application/json" \
  -d '{"input": "hi", "spec": {...}, "skills": {"disable": ["risky_skill"]}}'

Project config

project_config.yaml

Reference .kneo/config.yaml content showing project metadata, service defaults, runtime defaults, and per-environment policy enforcement overlays. Copy into .kneo/config.yaml to bootstrap a new project, or use:

kneo config init --name research-agent-demo
kneo config show

The schema and overlay rules are in project_config.md.

Helper Python

app_functions.py

Stub implementations for the tools and helpers referenced by research_agent.yaml (compress_history, web_search, webpage_reader). Ship-quality replacements would call real services; these just return formatted strings so the agent loop has something to do.

human_functions.py

Stub draft_report and publish_report used by the human-approval workflow. Same pattern as app_functions.py.

nested_functions.py

Stub nested-drafting steps (outline_section, expand_draft) and the top-level publish_report referenced by nested_workflow_human_approval.yaml. Same stub pattern.

guardrail_functions.py

Stub lookup_account tool for guardrails_complete.yaml; deliberately returns a record containing an SSN so the tool-stage guardrail has PII to redact.

Adapting an example

  1. Copy a spec into your project, e.g. cp examples/research_agent.yaml my_agent.yaml.
  2. Replace the model.provider / model.name with your provider, and add the corresponding env-var reference under your project secrets.
  3. Replace the tools with real ones — either Python functions registered through ToolRegistry or MCP servers (see extending.md).
  4. Validate against your target environment: 
    kneo spec validate my_agent.yaml --env prod
  5. Compile to confirm the workflow builds: 
    kneo spec compile my_agent.yaml
  6. Run locally before deploying: 
    kneo run --input "<prompt>" --target workflow my_agent.yaml

For deployment to a service, see deployment.md.

0.9.0 additions: resilience, handoff, stdio MCP

Three examples landed with the 0.9.0 reliability cut, each doubling as the executable regression for a fixed surface (they run offline in CI — no provider keys, no network):

  • handoff_workflow.yaml — a round_robin handoff: each participant takes one turn, then the run completes (status: completed, not the pre-0.9.0 max_iterations failure).

    kneo run examples/handoff_workflow.yaml \ --input "triage this incident report" --target workflow --json

  • resilient_workflow.yaml — per-step on_error: the fetch step always fails and falls back to cached_fetch; enrich always fails and is skipped (continue). The final report shows [cached] content — both policies actually executing. Semantics: run_lifecycle.md.

    kneo run examples/resilient_workflow.yaml \ --input "quarterly metrics" --target workflow --json

  • mcp_stdio_workflow.yaml — declarative MCP over stdio, backed by the bundled mcp_stdio_server.py (FastMCP, ships with the SDK's mcp dependency). The platform spawns the server as a subprocess on the tool's first call; the session lives on a dedicated event loop and is reused across calls. This is a runtime transport proof — the pre-0.9.0 stdio path failed on every invocation.

    kneo run examples/mcp_stdio_workflow.yaml \ --input "count the words in this sentence" --target agent --json

The retention: block in project_config.yaml now also shows every retention knob, including 0.9.0's idempotency_days.

0.10.0 additions: regression-showcase examples

Four examples landed with the 0.10.0 cut, each the executable proof of a cluster-0 fix. They run offline in CI (no provider keys, no network). Where an agent would need a provider to drive a tool call, the example proves enforcement at the seam instead (compile + wire + exercise, or a direct adapter call) — the YAML/script headers say which.

  • nested_workflow_human_approval.yaml (HIGH #1) — a kind: workflow nested drafting pipeline gated by a top-level human-approval step. The run blocks at approval and publishes only the approved draft; burying the human step inside the nested workflow is rejected at validation (E_STEP_WORKFLOW_NESTED_HUMAN), so an approval gate can never be silently bypassed.

    kneo run examples/nested_workflow_human_approval.yaml \ --input "the Q3 board report" --target workflow --json

  • guardrails_complete.yaml (HIGH #2) — a tool-stage guardrail that redacts PII (an SSN pattern) from a tool result. Before the fix a tool/workflow-stage guardrail validated, satisfied the production require_guardrails gate, deployed, and was never enforced. As of 0.11.0 every tool-stage action is enforced (a raising action aborts the run via GuardrailAbort) and workflow-stage guardrails are enforced per step, so the 0.10.0 validator-rejects (E_GUARDRAIL_ACTION_UNSUPPORTED / E_GUARDRAIL_STAGE_UNSUPPORTED) no longer fire.

  • redaction_demo.py — secret redaction across structured payloads, free text, and trace events. Pins the pluralized-key fix: api_keys / refresh_tokens are redacted while usage keys (input_tokens, max_tokens) survive.

    python examples/redaction_demo.py

  • custom_middleware_demo.py — a custom tool middleware plus the two adapter-hop fixes: a custom middleware's ToolResult.metadata reaches the SDK ToolCallContext.metadata, and the OpenTelemetry context survives the run_awaitable_sync worker-thread hop.

    python examples/custom_middleware_demo.py