Tutorial: writing a custom tool end-to-end¶

Build a simple custom tool, register it with kneo-serv, expose it in a YAML spec, run an agent that calls it, and verify how the call shows up in audit and trace events.

This tutorial walks the path from "I have a Python function" to "an agent uses it in production" so you can see every layer the call passes through.

For the recipe-style summary, see extending.md § 1. For the public-API surface, see implementation_map.md § tools/.

What we're building¶

A lookup_user tool that takes a user id and returns the user's email. We'll start with a function, register it as a tool, and run a small agent that uses it.

1 · Write the Python function¶

Create examples/my_tools.py (or anywhere on the import path):

# examples/my_tools.py
from typing import Any

USERS = {
    "u-001": "alice@example.com",
    "u-002": "bob@example.com",
    "u-003": "carol@example.com",
}


def lookup_user(args: dict[str, Any]) -> str:
    """Return the email for a user id, or 'unknown' if not present."""
    user_id = args.get("user_id")
    if not user_id:
        return "unknown: user_id is required"
    return USERS.get(user_id, f"unknown: {user_id}")

Two things to notice:

The handler signature is Callable[[dict[str, Any]], str]. Args arrive as a dict; the return value must be a string. The framework wraps it as a ToolResult.
The function should be deterministic and side-effect-free where possible. If it does I/O, plan for retries and timeouts (see environment.md § Runtime Reliability).

2 · Register the tool with `ToolRegistry`¶

Tools live in a ToolRegistry. The default service registry is built in service/factory.py; to add a custom tool you need a registry that includes it. The cleanest approach is to construct a custom registry and pass it into the platform manager.

Create examples/my_factory.py:

# examples/my_factory.py
from kneo_serv.platform import PlatformManager
from kneo_serv.spec import SpecCompiler
from kneo_serv.service.factory import (
    create_runtime_registry,
    create_tool_registry,
    create_persistence_stores,
)
from kneo_serv.tools import ToolDefinition

from examples.my_tools import lookup_user


def build_platform() -> PlatformManager:
    runtime_registry = create_runtime_registry()
    tool_registry = create_tool_registry(include_examples=True)
    tool_registry.register(
        ToolDefinition(
            name="lookup_user",
            description="Return the email for a user id.",
            parameters={
                "type": "object",
                "properties": {"user_id": {"type": "string"}},
                "required": ["user_id"],
            },
        ),
        lookup_user,
    )

    compiler = SpecCompiler(
        runtime_registry=runtime_registry,
        tool_registry=tool_registry,
    )
    run_state, continuation = create_persistence_stores()
    manager = PlatformManager(
        compiler=compiler,
        run_state_store=run_state,
        continuation_store=continuation,
    )
    manager.start_worker()
    return manager

The parameters schema is JSON Schema. The agent's tool-call planner sees this — be explicit about types, required fields, and enums. The guarded tool registry applies any policy from the spec on top of this definition, so you don't need to enforce auth here.

3 · Reference the tool from a spec¶

Create examples/lookup_agent.yaml:

version: v1

agent:
  name: directory-agent
  system_prompt: |
    You answer questions by looking up users with the lookup_user tool.
    Return concise, factual answers.
  model:
    provider: openai
    name: gpt-4o-mini
  strategy:
    type: react
    max_iterations: 4
  runtime_preferences:
    preferred_mode: bridge
    allowed_modes: [bridge]
  tools:
    include: [lookup_user]

workflow:
  type: sequential
  name: lookup-pipeline
  steps:
    - id: answer
      kind: agent
      ref: directory-agent

The tools.include list names tools the agent is allowed to call. Names must match what we registered in step 2. Validate the spec:

kneo spec validate examples/lookup_agent.yaml

If you see E_UNKNOWN_TOOL, the spec is referencing a name that isn't in the registry. Re-check step 2 — the registry must include lookup_user before SpecCompiler runs.

4 · Run the agent¶

When using a custom factory, drive the agent through Python rather than the bare kneo run command — the CLI's default platform manager does not include your custom tool.

# examples/run_lookup.py
from examples.my_factory import build_platform


def main() -> None:
    manager = build_platform()
    result = manager.run_from_spec(
        input_text="What's the email for u-002?",
        spec_path="examples/lookup_agent.yaml",
        target="workflow",
    )
    print(f"run_id={result.run_id}")
    print(result.output_text)


if __name__ == "__main__":
    main()

Run it:

export OPENAI_API_KEY=sk-...
python -m examples.run_lookup

Expected output (model wording will vary):

The email for u-002 is bob@example.com.

If the agent didn't call the tool, increase max_iterations or refine the system prompt to instruct tool use.

5 · Inspect the call in trace and audit¶

The tool call shows up in the run trace. Capture the run_id the script prints and pass it to kneo runs trace:

RUN_ID=$(python -m examples.run_lookup | sed -n 's/^run_id=//p')
kneo runs trace "$RUN_ID"

Look for tool_call_started and tool_call_completed events with tool_name: "lookup_user". Tool arguments and results are not in the trace by default — they're redacted at write time. To capture them in OpenTelemetry spans for a specific deployment, opt in with KNEO_SERV_OTEL_RECORD_ARGUMENTS=true and KNEO_SERV_OTEL_RECORD_RESULTS=true (only after a data-classification review; see environment.md § Observability).

The audit log records the run (not individual tool calls) at run.created / run.cancelled / run.continued. Tool arguments are not persisted in audit events under any flag — see production_readiness_review.md § Audit Payload Review.

6 · Lock down the tool with policy¶

For production, restrict what the agent can do with your tool. Add to the spec under agent:

agent:
  # ... existing fields ...
  policies:
    tool:
      allow:
        - lookup_user
      deny: []
      network: false
      filesystem: false
      shell: false

The guarded registry blocks every tool not in allow, and diagnostics check that registered tools don't claim capabilities (network, filesystem, shell) inconsistent with the policy.

Validate and re-run; the policy is enforced at call time. A blocked call surfaces as a tool_policy_denied trace event and never reaches your handler.

7 · Ship the tool to the service¶

Two patterns:

In-process custom factory. Replace service.factory with your own create_default_platform_manager() that registers the tool. Run kneo service serve from a deployment that imports your factory. This is the common shape; the Docker image is a thin wrapper around service.app:create_app(configure_default_manager=True), and you can override the manager in your own entrypoint.
Custom server. Construct the FastAPI app yourself with create_app(configure_default_manager=False) and call kneo_serv.service.dependencies.set_platform_manager(your_manager) before serving. Useful when you want to register multiple tools, or combine custom auth + custom tools.

# examples/serve_lookup.py
import uvicorn
from kneo_serv.service.app import create_app
from kneo_serv.service.dependencies import set_platform_manager

from examples.my_factory import build_platform

if __name__ == "__main__":
    set_platform_manager(build_platform())
    uvicorn.run(create_app(), host="127.0.0.1", port=8000)

python -m examples.serve_lookup
curl -sf http://127.0.0.1:8000/livez

Now the same lookup_user tool is reachable from any spec that references it, including service-backed CLI calls and HTTP POST /v1/runs.

Common pitfalls¶

Symptom	Likely cause
`E_UNKNOWN_TOOL` on `kneo spec validate`	Custom tool isn't in the registry the CLI is using; use the Python entrypoint from § 4.
`ValueError: Tool 'X' has no implementation`	`ToolDefinition.name` doesn't match the spec's `tools.include` entry.
Agent never calls the tool	System prompt doesn't mention it, or `max_iterations` is too low.
Tool calls succeed but content is missing in trace	Expected — tool args/results are redacted by default.
`403 Forbidden` when calling through service	API key is missing the `runs:write` scope; see troubleshooting.md § 4.2.

Next¶

More examples and orchestration patterns: examples.md.
Custom MCP servers (similar pattern, externally hosted tools): extending.md § 2.
Custom middleware around tool calls (rate-limiting, logging): extending.md § 3.