kodelet-sdk

Run Kodelet sessions and write tools, commands, and event handlers in Python.

Quick start

uv add kodelet-sdk

An executable extension registers handlers and serves them over stdio:

from kodelet_sdk import BaseModel, Extension, ToolContext, ToolExecutionResult

ext = Extension(name="weather", version="0.1.0")


class WeatherInput(BaseModel):
    location: str


@ext.tool("get_weather", description="Get weather", input_schema=WeatherInput)
async def get_weather(input: WeatherInput, ctx: ToolContext) -> ToolExecutionResult:
    return {"content": f"Weather for {input.location}"}


@ext.on("session.start")
async def session_start(event, ctx):
    ctx.log.info("extension started")


if __name__ == "__main__":
    ext.run_sync()

Agent sessions

Client uses your configured kodelet CLI to connect to a daemon and runner. Set server and runner on Client to select them explicitly. Provider credentials stay on the daemon; session cwd refers to the runner's workspace.

Run the following snippets inside an async function. The inline-extension example below includes a complete script.

from kodelet_sdk import Client

client = Client()
session = await client.create_session()
response = await session.run_and_wait(message="what is the meaning of life?")

print(response.content)
await client.close()

Streaming

Choose a profile and subscribe to session events:

from kodelet_sdk import Client

client = Client()
session = await client.create_session(
    profile="work",  # A model profile configured on your daemon.
    max_turns=4,
    streaming=True,
)

session.on(
    "assistant.message_delta",
    lambda event: print(event.data.deltaContent, end="", flush=True),
)
session.on(
    "tool.update",
    lambda event: print(f"partial {event.data.toolCallId}: {event.data.result}"),
)
session.on(
    "tool.result",
    lambda event: print(f"final {event.data.toolCallId}: {event.data.result}"),
)

response = await session.run_and_wait(message="help me choose an approach")
print("\nfinal:", response.content)
await client.close()

tool.update replaces the previous snapshot for that tool call; it is not a delta. Listeners receive every update, while response.events keeps only the latest snapshot and the final result.

Session options include:

Inline settings cannot include provider secrets, endpoints, or local prompt paths.

Inline extensions

Pass extensions=[ext] to expose local Python callbacks as agent tools. This calculator runs in your Python process; the agent runs on the selected daemon and runner.

import asyncio

from kodelet_sdk import BaseModel, Client, Extension


ext = Extension(name="calculator", version="0.1.0")


class CalculatorInput(BaseModel):
    a: int
    b: int


@ext.tool("calculator", description="Add two integers", input_schema=CalculatorInput)
async def calculator(input: CalculatorInput) -> str:
    return str(input.a + input.b)


async def main() -> None:
    client = Client()
    try:
        session = await client.create_session(extensions=[ext])
        response = await session.run_and_wait("Use calculator to add 123 and 456")
        print(response.content)
    finally:
        await client.close()


if __name__ == "__main__":
    asyncio.run(main())

Inline extensions require ACP session-extension protocol v1; older hosts fail with a clear error. On resume, reattach extensions in the same order. Callbacks are not saved or replayed, and pending work is cancelled when the channel or session closes.

Host calls such as ctx.update() and ctx.children go to the runner. File, process, and storage helpers remain local. Legacy extension_transport="unix" and "tcp" options are accepted but ignored; ACP manages the connection.

Steering

Call session.steer() after an event confirms the run is active:

import asyncio

run_active = asyncio.Event()
session.once("assistant.thinking_start", lambda _event: run_active.set())
run_task = asyncio.create_task(session.run_and_wait("Review the persistence implementation"))

await run_active.wait()
steered = await session.steer("Also check transaction boundaries")
response = await run_task

Steering requires host support. injected means guidance was queued, not necessarily consumed. If the turn has just ended, the result is promptRequired; steering never starts a new turn automatically.

Extension registration

Create an Extension(name=..., version=...), then register synchronous or asynchronous handlers:

Tool results and progress

Return a string or a mapping with content and optional data and error fields. Use data["presentation"] to customize the displayed result without changing what the model receives:

from kodelet_sdk import ToolExecutionResult, ToolPresentation

presentation: ToolPresentation = {
    "summary": "Found 2 matches",
    "body": "- `src/api.py`\n- `tests/test_api.py`",
    "format": "markdown",
}
result: ToolExecutionResult = {
    "content": "Found 2 matches.",
    "data": {"presentation": presentation},
}

summary is required; body is optional and supports text or markdown. Hosts may sanitize or truncate display content.

For live progress, call ctx.update(). Updates replace earlier snapshots; only the final return value is persisted and sent to the model:

@ext.tool("search", description="Search a project")
async def search(input, ctx: ToolContext) -> str:
    await ctx.update("Searching code", {"filesScanned": 12})
    return "Search complete"

ctx.update() is a no-op on hosts without progress support. For multi-step tasks, TaskProgress tracks activities and can attach to session events; await progress.finish(...) ends tracking and detaches listeners.

Commands and events

from kodelet_sdk import (
    CommandContext,
    CommandResult,
    EventContext,
    ToolCallEvent,
    ToolUpdateEvent,
)


@ext.command("doctor", description="Check health", input_schema=WeatherInput)
async def doctor(input: WeatherInput, ctx: CommandContext) -> CommandResult:
    return {"action": "respond", "response": ctx.input["commandName"]}


@ext.on("tool.call")
def approve(event: ToolCallEvent, ctx: EventContext):
    return {"message": event.tool.name}


@ext.on("tool.update")
def sanitize_partial_output(event: ToolUpdateEvent, ctx: EventContext):
    return {"output": event.tool.output}

Commands return one of:

• {"action": "pass"} — let another route handle the command.
• {"action": "respond", "response": "..."} — respond directly to the user.
• {"action": "runAgent", "prompt": "..."} — run the agent with a replacement prompt; optional display controls the visible user message.

If you sanitize tool.result, apply the same policy to tool.update so partial output is also safe to display. Without an update handler, Kodelet suppresses partial output for result-subscribing extensions.

Keyboard shortcuts

Shortcuts run in local native kodelet chat sessions:

from kodelet_sdk import ShortcutContext


@ext.shortcut("ctrl+alt+r", description="Refresh project context")
async def refresh(ctx: ShortcutContext) -> None:
    await ctx.ui.notify("Project context refreshed")

Supported chords are ctrl+<letter>, alt+<letter-or-digit>, ctrl+alt+<letter>, and f1–f12 (ASCII, case-insensitive). ctrl+i and ctrl+m are excluded because terminals treat them as Tab and Enter. Reserved host bindings take precedence.

Handlers can return {"action": "submit", "message": "/dictate"} when the host supports shortcut submission.

Schemas and templates

The SDK re-exports Pydantic types and provides Jinja2 rendering:

from kodelet_sdk import BaseModel, Field, render_template


class ReviewInput(BaseModel):
    target: str = Field(min_length=1)


assert render_template("Review {{ target }}", {"target": "main"}) == "Review main"

Pydantic schemas validate inputs before handlers run and generate JSON Schema for the host. Invalid command inputs return {"action": "pass"}. Raw JSON Schema mappings are also accepted, but do not perform local input validation.

Context helpers

Handlers receive ctx with call metadata and these helpers:

• ctx.storage.read_text/write_text/read_json/write_json(...) for extension data files.
• ctx.path.resolve_workspace_path(...) and ctx.path.relative_to_workspace(...).
• ctx.fs.exists/read_text/write_text/list(...) for workspace file access.
• ctx.process.exec(...) and ctx.process.spawn(...) for async process execution.
• ctx.env.get(...) for environment access.
• ctx.log.debug/info/warn/error(...) for JSON logs to stderr.
• ctx.children.start(...) for delegated agent work.
• await ctx.acquire_background_task(...) to keep runtime resources alive after a handler returns.
• ctx.ui.input/confirm/select/notify(...) for host UI reverse-RPC calls.
• ctx.ui.append_transcript(...), ctx.ui.set_widget(...), and ctx.ui.open_surface(...) for capability-gated persistent native-TUI content.

Child agents

Use ctx.children for agent work inside a tool, rather than creating a nested Client. Register a preset on the extension, then start a child with that preset:

class TaskInput(BaseModel):
    task: str


ext.register_profile({
    "name": "code_search",
    "options": {
        "allowed_tools": ["file_read", "grep_tool", "glob_tool"],
        "no_extensions": True,
        "no_skills": True,
        "enable_fs_search_tools": True,
        "max_turns": 3,
    },
})


@ext.tool("code_search", description="Search the repository", input_schema=TaskInput)
async def code_search(input: TaskInput, ctx: ToolContext) -> str:
    child = await ctx.children.start(profile="code_search", message=input.task)
    result = await child.wait(on_event=lambda event: ctx.update(event.get("text") or event["kind"]))
    return result["output"]

• read() checks progress; cancel() cancels only that child. Cancelling wait() also cancels the child.
• Children start with fresh context. Use context_mode="fork" inside the active tool handler to copy the parent's history.
• Use resume=child.conversation_id for a follow-up. It creates a new run; old handles still target the old run. Resume cannot be combined with fork.
• child.steer(message) queues guidance without starting a turn. Like session steering, it returns injected or promptRequired.
• Child settings cannot expand the parent's permissions or limits. system_prompt_path in a preset is resolved on the runner.

Start and steer calls do not retry automatically. Supply a stable request_id when you need to reconcile an uncertain submission; reuse it only for the same input. A disconnected start may already be running, so keep its ID and any lease until cancellation or cleanup is confirmed.

For a history snapshot without execution, use ctx.fork_conversation(). It raises ConversationForkUnavailableError when unavailable. Client.create_session(inherit_context=...) is unsupported.

Background work

Children normally end with their tool handler. To let work outlive it, acquire a lease with await ctx.acquire_background_task(...) and pass lease=lease when first starting the child inside the tool. Close the lease after the work and final updates finish.

Runner leases last at most one hour and end on release, cancellation, or runner/extension shutdown. They keep resources alive, not Python task state. Resuming after a restart requires a new authorized tool call; saved IDs alone are insufficient.

User input

Executable extensions use the host's UI. For inline extensions, pass ui={"select": handler, ...} to create_session() to handle requests locally. A local handler needs no terminal unless it uses one; requests without a handler depend on the runner's UI support.

from kodelet_sdk import UIInputRequest, UISelectRequest

input_request: UIInputRequest = {"title": "Branch name", "required": True}
select_request: UISelectRequest = {"title": "Mode", "options": ["fast", "thorough"]}

branch = await ctx.ui.input(input_request)
mode = await ctx.ui.select(select_request)

Persistent TUI content

Use transcript entries, widgets, and surfaces for longer-lived UI. Widget and surface IDs are scoped to the originating conversation.

await ctx.ui.append_transcript({"title": "Drawing saved", "message": "./drawing.png"})
await ctx.ui.set_widget("status", ["Indexing repository…"])

surface = await ctx.ui.open_surface(
    {
        "id": "preview",
        "initialLines": ["Loading…"],
        "width": "75%",
        "height": "80%",
        "anchor": "center",
    }
)

surface.on_resize(
    lambda event: surface.update([f"Surface size: {event['width']}×{event['height']}"])
)
surface.update(["Preview ready"])
await surface.close()
await ctx.ui.set_widget("status", None)  # Remove the widget.

append_transcript() and set_widget() are no-ops without host support; open_surface() raises RuntimeError. Surface updates replace existing content. If surface.close() fails, the handle remains valid for retry.

Runtime behavior

• Requests run concurrently. Cancellation or disconnection raises asyncio.CancelledError in async handlers; late request-scoped UI calls and updates are rejected.
• Persistent widgets and surfaces can outlive the handler that opened them, but remain scoped to their conversation.
• ACP messages are limited to 64 MiB each. Pipe failures and oversized messages fail pending requests and stop the subprocess.

Testing extensions

Test handlers without a subprocess:

from kodelet_sdk import Extension, create_test_harness


async def test_tool():
    ext = Extension(name="example")

    @ext.tool("echo", description="Echo", input_schema={"type": "object"})
    async def echo(input, ctx):
        return {"content": input["text"]}

    harness = await create_test_harness(ext)
    result = await harness.execute_tool({"name": "echo", "input": {"text": "hi"}})
    assert result == {"content": "hi"}

Examples

Run the inline calculator from a checkout with a configured Kodelet daemon:

uv run -s examples/sdk/inline-extension-session

Other examples:

• examples/sdk/basic-agent-session — one prompt and its final response.
• examples/sdk/streaming-agent-session — live assistant and tool output.
• examples/review/kodelet-extension-review — review command extension.
• examples/workspace/kodelet-extension-workspace — workspace helper/policy extension.