kindgi

Kindgi™ for Python.

Write a pack's tools and guardrail checks in Python:

from kindgi import ToolContext, tool

A tool that is one HTTP request needs no code: http_tool(...).

Agents and flows are data (Agent, Flow). python -m kindgi.pack index writes the pack's index; python -m kindgi.pack serve runs its code for a Kindgi runtime (kindgi dev does both for you).

MAX_STORED_FILES_PER_CALL

MAX_STORED_FILES_PER_CALL = 10

Signature

MAX_STORED_FILE_BYTES

MAX_STORED_FILE_BYTES = 10000000

Signature

Agent

class Agent(
    id: str,
    version: str,
    name: str,
    instructions: str | Mapping[str, str],
    capabilities: Sequence[Mapping[str, Any]] = (),
    tools: Sequence[Tool[..., Any] | Mapping[str, Any]] = (),
    retrieval: Sequence[Mapping[str, Any]] = (),
    guardrails: Sequence[Guardrail | str] = (),
    parameters: Sequence[Mapping[str, Any]] = (),
    budget: Mapping[str, Any] | None = None,
    preferred_provider: str | None = None,
    preferred_model: str | None = None,
    description: str | None = None,
    tags: Sequence[str] | None = None,
    conversation_policy: Mapping[str, Any] | None = None,
    output: Any = None,
    tool_errors: Mapping[str, Any] | None = None,
    parallel_tool_calls: bool | None = None,
    settings: Sequence[Mapping[str, str]] = (),
    model_settings: Mapping[str, str] | None = None,
    memory: Mapping[str, Any] | None = None,
    module: str = '',
)

Signature

An agent: instructions, capabilities, the tools it may call. Pure data.

tools takes Tool objects (pinned to their version) or {"id", "version"} refs (version a semver range); guardrails takes Guardrail objects or ids; output a model class / type / JSON Schema for typed output, or the full {"schema", "name", "maxRepairs"} object.

instructions is the system prompt, or a prompt block by range ({"prompt": "acme.intake-prompt", "version": "^1.0.0"}) whose template and parameters are used instead. settings lists settings blocks by range ({"id", "version"}), which tools read as ctx.settings[id]; model_settings names a model-settings block. The block versions are pinned when the agent version is published.

CallCancelled

class CallCancelled(reason: CancelReason)

Signature

Raised by Cancellation.raise_if_cancelled once the call is over.

Cancellation

class Cancellation()

Signature

Fires when the caller gives up on the call: its deadline passed, or it disconnected.

The Python counterpart of ToolContext.abortSignal. An async handler is also cancelled the asyncio way (CancelledError at its next await); a sync handler runs in a worker thread and should check cancelled (or wait on it) between slow steps. Thread-safe.

Cancellation.cancelled

Cancellation.reason

Cancellation.raise_if_cancelled()

raise_if_cancelled() -> None

Signature

Cancellation.wait()

wait(timeout: float | None = None) -> bool

Signature

Block until cancelled or timeout seconds pass; True when cancelled.

Cancellation.wait_async()

wait_async() -> CancelReason

Signature

Resolve once cancelled.

Cancellation.cancel()

cancel(reason: CancelReason) -> None

Signature

Fire (idempotent; the first reason wins).

CheckResult

class CheckResult(**data: Any)

Signature

A check's verdict. passed=False with a reason fires the guardrail's action.

Fields:

• passed: bool
• reason: str | None (optional)
• judge_response: str | None (judgeResponse on the wire) (optional)
• attributes: dict[str, Any] | None (optional)

CheckResult.to_wire()

to_wire() -> dict[str, Any]

Signature

ConnectionAccount

class ConnectionAccount(id: str, label: str)

Signature

Whose account a connection is at the app.

ConnectionRejected

class ConnectionRejected(connection: str, message: str | None = None)

Signature

The app refused a connection's token: an HTTP 401 from its API.

Raise it with the tool's name for the connection (its needs_spec["connections"] key), or raise your own error from it. The call answers connection-rejected, and the runtime renews that token once and runs the call again; it never removes the connection. Say what the app said, never the token.

DefinitionError

class DefinitionError(*args, **kwargs)

Signature

A primitive is declared wrong — raised where it is declared.

FileReadAnswer

class FileReadAnswer(data: bytes | None = None, not_read: NotReadReason | None = None)

Signature

What a read answers: data, the file's bytes as it was stored, or not_read, why there are none.

FileStore

class FileStore(_url: str, _token: str, stores: bool = False, reads: bool = False)

Signature

A call's files (ToolContext.files): store for a tool that declares stores_files=True, on a call with an idempotency key, and read for one that declares reads_files=True. stores and reads say which the call has; the other raises FileStoreError.

Storing the same bytes again in the same call, a re-run of it included, answers the first file and stores nothing new. A call stores at most MAX_STORED_FILES_PER_CALL files.

FileStore.store()

store(*, name: str, content_type: str, data: bytes) -> FileStoreAnswer

Signature

Store one file, from a sync handler. name is its name as its source gives it (an empty one is stored as unnamed); content_type the type its source declares, which the runtime checks the bytes against. Raises FileStoreError when the runtime refuses the store or can't be reached.

FileStore.store_async()

store_async(*, name: str, content_type: str, data: bytes) -> FileStoreAnswer

Signature

store, from an async handler.

FileStore.read()

read(artifact_id: str) -> FileReadAnswer

Signature

Read a stored file's bytes, from a sync handler, by its file value's artifactId: a file the call's run may be shown, such as one in the run's input, in its conversation, or one a tool stored in it. Answers the bytes it was stored with, never a copy made for a model, as data, or why there are none, as not_read. Raises FileStoreError when the runtime refuses the read or can't be reached.

FileStore.read_async()

read_async(artifact_id: str) -> FileReadAnswer

Signature

read, from an async handler.

FileStoreAnswer

class FileStoreAnswer(
    file: Mapping[str, Any] | None = None,
    not_stored: NotStoredReason | None = None,
)

Signature

What a store answers: file, the stored file's value, or not_stored, why nothing was.

Return file in the result's top-level files ({"files": [answer.file]}), and an agent is shown the file. For not_stored, return the reason as the file's notShown, so the agent reads a line naming it.

FileStoreError

class FileStoreError(*args, **kwargs)

Signature

Raised by FileStore's methods when the runtime refuses the store or the read or can't be reached, or when the call has no such method (stores, reads).

Let it propagate: the call fails as any exception fails it.

Flow

class Flow(
    id: str,
    version: str,
    nodes: Sequence[Mapping[str, Any]],
    edges: Sequence[Mapping[str, Any]],
    name: str | None = None,
    description: str | None = None,
    max_parallelism: int | None = None,
    metadata: Mapping[str, Any] | None = None,
    output: Mapping[str, Any] | None = None,
    module: str = '',
)

Signature

A flow: nodes and edges as data (flow.schema.json). A node's ref may be a Tool.

Guardrail

class Guardrail(
    id: str,
    check: CheckFn,
    check_id: str,
    kind: str,
    action: Mapping[str, Any],
    name: str | None = None,
    severity: str | None = None,
    scope: Mapping[str, Any] | None = None,
    config: Mapping[str, Any] | None = None,
    config_schema: dict[str, Any] | None = None,
    config_type: Any = None,
    trace_is_model: bool = True,
    is_async: bool = False,
    sandbox: str | None = None,
    limits: Mapping[str, Any] | None = None,
    network: Mapping[str, Any] | None = None,
    judge_capabilities: Mapping[str, Any] | None = None,
    module: str = '',
    _config_adapter: TypeAdapter[Any] | None = None,
)

Signature

A guardrail whose check is this module's function. Calling it calls the check.

Guardrail.parse_config()

parse_config(value: Mapping[str, Any]) -> Any

Signature

Guardrail.parse_trace()

parse_trace(value: Any) -> Any

Signature

ModelCallRecord

class ModelCallRecord(**data: Any)

Signature

Fields:

• provider_id: str (providerId on the wire)
• model: str
• prompt_tokens: int (promptTokens on the wire)
• completion_tokens: int (completionTokens on the wire)
• at: str

PauseAnswer

class PauseAnswer(timeout_ms: int | None = None, name: str | None = None)

Signature

What ctx.pause() returns. A pausing tool's handler returns it as its result.

PauseNotAvailable

class PauseNotAvailable(*args, **kwargs)

Signature

Raised by ToolContext.wait_token and ToolContext.pause when the call has no wait token.

The pack service answers the call with pause-not-available.

RunTrace

class RunTrace(**data: Any)

Signature

The run a check evaluates: its output, tool calls and results, model calls.

Fields:

• run_id: str (runId on the wire)
• tenant_id: str (tenantId on the wire)
• project_id: str | None (projectId on the wire) (optional)
• org_id: str | None (orgId on the wire) (optional)
• agent_id: str | None (agentId on the wire) (optional)
• flow_id: str | None (flowId on the wire) (optional)
• output: str | None (optional)
• tool_calls: list[kindgi.pack.trace.ToolCallRecord] (toolCalls on the wire) (optional)
• tool_results: list[kindgi.pack.trace.ToolResultRecord] (toolResults on the wire) (optional)
• model_calls: list[kindgi.pack.trace.ModelCallRecord] (modelCalls on the wire) (optional)
• user_input: str | None (userInput on the wire) (optional)
• retrieved_fact_ids: list[str] | None (retrievedFactIds on the wire) (optional)
• conversation_id: str | None (conversationId on the wire) (optional)
• turn_number: int | None (turnNumber on the wire) (optional)
• total_cost_usd: float | None (totalCostUsd on the wire) (optional)
• duration_ms: float | None (durationMs on the wire) (optional)
• mode: Literal['ci', 'runtime'] (optional)
• attributes: dict[str, Any] | None (optional)

Tool

class Tool(
    id: str,
    handler: Callable[P, R],
    description: str,
    version: str | None,
    input_schema: dict[str, Any],
    output_schema: dict[str, Any],
    input_type: Any,
    output_type: Any,
    takes_context: bool,
    is_async: bool,
    effects: Sequence[Mapping[str, Any]] = (),
    mutating: bool | None = None,
    needs_approval_when_unattended: bool | None = None,
    result_is_outside_data: bool | None = None,
    needs: Sequence[Mapping[str, Any]] | None = None,
    needs_spec: Mapping[str, Any] | None = None,
    sandbox: str | None = None,
    limits: Mapping[str, Any] | None = None,
    network: Mapping[str, Any] | None = None,
    spec: Mapping[str, Any] | None = None,
    pauses: bool = False,
    stores_files: bool = False,
    reads_files: bool = False,
    module: str = '',
    _input_adapter: TypeAdapter[Any] | None = None,
    _output_adapter: TypeAdapter[Any] | None = None,
)

Signature

A tool: its manifest fields plus the handler. Calling it calls the handler.

Tool.parse_input()

parse_input(value: Any) -> Any

Signature

The handler's argument for a wire input (raises pydantic ValidationError).

Tool.dump_output()

dump_output(value: Any) -> Any

Signature

The wire form of the handler's return value.

Tool.invoke_args()

invoke_args(value: Any, ctx: ToolContext) -> tuple[Any, ...]

Signature

ToolCallRecord

class ToolCallRecord(**data: Any)

Signature

Fields:

• tool_id: str (toolId on the wire)
• tool_name: str (toolName on the wire)
• arguments: dict[str, Any] (optional)
• at: str

ToolConnection

class ToolConnection(
    connection_id: str,
    app: str,
    app_name: str,
    access_token: str,
    account: ConnectionAccount,
    expires_at: str | None = None,
)

Signature

A connection resolved for one call: an account at another app, with a short-lived token.

What a tool reads from ctx.connections["mail"]. The token is never in its repr.

ToolConnection.from_wire()

from_wire(value: Any) -> ToolConnection | None

Signature

The connection for one protocol v2 ctx.connections entry; None when it isn't one.

ToolContext

class ToolContext(
    tenant_id: str,
    run_id: str,
    request_id: str | None = None,
    project_id: str | None = None,
    org_id: str | None = None,
    env: Mapping[str, Any] = <factory>,
    secrets: Mapping[str, Any] = <factory>,
    config: Mapping[str, Any] = <factory>,
    settings: Mapping[str, Mapping[str, Any]] = <factory>,
    cancellation: Cancellation = <factory>,
    log: Logger = Logger(subsystem='root', level='error'),
    idempotency_key: str | None = None,
    connections: Mapping[str, ToolConnection] = <factory>,
    files: FileStore | None = None,
    _wait_token: WaitToken | None = None,
)

Signature

Per-call context: who the call is for, its resolved env/secrets/config, its cancellation.

ToolContext.wait_token

The call's wait token, for a tool that declares pauses=True: hand its url to the system that will answer, then return ctx.pause(). The run waits until that system posts its answer to the link, or until the pause's time passes; the answer, valid against the tool's output schema, becomes the call's result. A re-run after a crash gets a new link to the same wait, and either link completes it: hand on the new one only if the old one never went out, and dedupe on idempotency_key.

Raises PauseNotAvailable when the call has none: the tool doesn't declare pauses=True, or the runtime that called it can't pause (before pack protocol 2.8.0). Read it before handing anything on, so the tool fails first.

ToolContext.pause()

pause(timeout_ms: int | None = None, name: str | None = None) -> PauseAnswer

Signature

The answer that makes the run wait for the system the tool handed wait_token to.

timeout_ms: how long the run waits, from 1 to MAX_PAUSE_TIMEOUT_MS (24 days); 24 hours when None. name: what it waits for, in a few words. Raises PauseNotAvailable as wait_token does.

ToolContext.for_test()

for_test(
    tenant_id: str = 'tenant-test',
    run_id: str = 'run-test',
    *,
    wait_token: WaitToken | None = None,
    tool: Tool[Any, Any] | None = None,
    **kwargs: Any,
) -> ToolContext

Signature

A context for calling a handler directly in a unit test. Pass wait_token to test a pausing tool.

Pass tool= to fill env from the tool's needs_spec["env"] the way the runtime does: every env name the tool declares, from the env= the test offers or the schema's string default, each checked against its schema. A name the tool doesn't declare is dropped. secrets= pass through unchecked (the runtime resolves them).

ToolContext.from_wire()

from_wire(
    ctx: Mapping[str, Any],
    cancellation: Cancellation,
    log: Logger = Logger(subsystem='root', level='error'),
    *,
    pauses: bool = False,
    stores_files: bool = False,
    reads_files: bool = False,
) -> ToolContext

Signature

The context for a protocol v2 ctx (tenantId, runId, …). Its waitToken is kept only for a tool that pauses, and its toolFiles, as files, only for a tool that stores_files (on a call with an idempotencyKey) or reads_files, with what it declares.

ToolResultRecord

class ToolResultRecord(**data: Any)

Signature

Fields:

• tool_call_id: str (toolCallId on the wire)
• output: Any (optional)
• at: str

WaitToken

class WaitToken(id: str, url: str)

Signature

A call's wait token: its id, and the link the other system posts its answer to.

guardrail

guardrail(
    *,
    id: str,
    kind: str = 'zero-llm',
    on_violation: str | None = None,
    action: Mapping[str, Any] | None = None,
    name: str | None = None,
    severity: str | None = None,
    scope: Mapping[str, Any] | None = None,
    check_id: str | None = None,
    config: Mapping[str, Any] | None = None,
    config_type: Any = None,
    sandbox: str | None = None,
    limits: Mapping[str, Any] | None = None,
    network: Mapping[str, Any] | None = None,
    judge_capabilities: Mapping[str, Any] | None = None,
) -> Callable[[CheckFn], Guardrail]

Signature

Declare a guardrail and its check: (config, trace) -> CheckResult | dict | bool.

on_violation is the action's name ("halt", "retry", "escalate", …); pass action= for the full object ({"on-violation": "retry", "retry": {"maxAttempts": 2}}). The runtime acts on "halt": the turn fails with guardrail-violation. It acts on "retry" too: the answer goes back to the model with the check's reason, for a new one, until the guardrail has failed maxAttempts answers (the first one included; 2 when left out), and then the turn fails the same way. Any other action is recorded on the turn's violations and the turn completes; in 0.1 nothing escalates or calls a compensating tool. config= is what the check is configured with, keyed as on the wire (a model's aliases); it is checked against the config type here. Without it the check gets {} — its defaults. The config type comes from the check's first annotation or config_type=. A kind="llm-judge" guardrail is judged by a model, with its config ({"rubric": ...}), and its check is never called. judge_capabilities= is what that model is routed by ({"needs": [{"feature": "structured-output"}]}); an llm-judge guardrail needs it.

http_tool

http_tool(
    *,
    id: str,
    description: str,
    input: Any,
    output: Any,
    method: HttpMethod,
    url_template: str,
    headers: Mapping[str, str] | None = None,
    authorization: Mapping[str, Any] | None = None,
    request_body: Mapping[str, Any] | None = None,
    timeout_ms: int | None = None,
    parse_json: bool | None = None,
    success_status: tuple[int, int] | None = None,
    version: str | None = None,
    effects: Sequence[Mapping[str, Any]] = (),
    mutating: bool | None = None,
    needs_approval_when_unattended: bool | None = None,
    result_is_outside_data: bool | None = None,
) -> Tool[..., Any]

Signature

Declare a tool that is one HTTP request — no handler: the Kindgi runtime makes the call.

The Python counterpart of defineTool({ spec: { kind: 'http', ... } }); the pack's index carries the spec. {name} placeholders in url_template (and in a text request body's template) are filled from the input's fields, URL-encoded. authorization and request_body take the spec's JSON shape:

authorization={"kind": "bearer", "secretRef": {"envName": "local", "name": "ACME_TOKEN"}}
request_body={"kind": "json-input"}

The runtime resolves the secret on every call (local: the pack's .env under kindgi dev). A GET that changes nothing is mutating=False. Calling the tool in Python raises: it runs in Kindgi.

tool

tool(
    *,
    id: str,
    version: str | None = None,
    description: str | None = None,
    input: Any = None,
    output: Any = None,
    effects: Sequence[Mapping[str, Any]] = (),
    mutating: bool | None = None,
    needs_approval_when_unattended: bool | None = None,
    result_is_outside_data: bool | None = None,
    needs: Sequence[Mapping[str, Any]] | None = None,
    needs_spec: Mapping[str, Any] | None = None,
    sandbox: str | None = None,
    limits: Mapping[str, Any] | None = None,
    network: Mapping[str, Any] | None = None,
    pauses: bool = False,
    stores_files: bool = False,
    reads_files: bool = False,
) -> Callable[[Callable[P, R]], Tool[P, R]]

Signature

Declare a tool. The handler is (input) or (input, ctx: ToolContext), sync or async.

version defaults to the pack's version; description to the docstring. mutating=False declares it read-only: it runs in a dry run, an agent's approval gates don't ask before it by default, and an agent runs a step's calls to read-only tools that come one after another at the same time, up to 8 at once (not one that pauses or declares connections, since it may wait). needs_approval_when_unattended=True makes each call wait for a person's approval in a run no person started (a webhook, a schedule, an event, or a service account's key). result_is_outside_data=True says the tool's results are text from outside that nobody checked (a mail someone sent, a page a search found): an agent reads each as data, not instructions.

pauses=True lets a call wait for another system's answer: the handler hands ctx.wait_token.url to that system and returns ctx.pause(); the answer that system posts becomes the call's result. Annotate the return as Answer | PauseAnswer: the output schema is Answer's.

stores_files=True lets a call store files it fetches, such as a mail's attachments: ctx.files.store(name=…, content_type=…, data=…) answers the stored file's value, which the handler returns in its result's top-level files, and an agent is shown the file.

reads_files=True lets a call read stored files' bytes, such as a PDF an agent was given: ctx.files.read(artifact_id) answers the bytes of a file the call's run may be shown, as it was stored, or why there are none.