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).

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,
    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.

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,
    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,
    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>,
    _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,
) -> ToolContext

Signature

The context for a protocol v2 ctx (tenantId, runId, …). Its waitToken is kept only for a tool that pauses.

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,
) -> 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}}). In 0.1 the runtime acts only on "halt": the turn fails with guardrail-violation. Any other action is recorded on the turn's violations and the turn completes; nothing runs it again, 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=.

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,
) -> 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, and an agent's approval gates don't ask before it by default. 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.