# kindgi

> Tools, guardrails, agents and flows for a Python pack.

Kindgi™ for Python.

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

```plaintext
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`

```python signature
MAX_STORED_FILES_PER_CALL = 10
```

## `MAX_STORED_FILE_BYTES`

```python signature
MAX_STORED_FILE_BYTES = 10000000
```

## `Agent`

```python signature
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 = '',
)
```

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`

```python signature
class CallCancelled(reason: CancelReason)
```

Raised by `Cancellation.raise_if_cancelled` once the call is over.

## `Cancellation`

```python signature
class Cancellation()
```

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()`

```python signature
raise_if_cancelled() -> None
```

### `Cancellation.wait()`

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

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

### `Cancellation.wait_async()`

```python signature
wait_async() -> CancelReason
```

Resolve once cancelled.

### `Cancellation.cancel()`

```python signature
cancel(reason: CancelReason) -> None
```

Fire (idempotent; the first reason wins).

## `CheckResult`

```python signature
class CheckResult(**data: Any)
```

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()`

```python signature
to_wire() -> dict[str, Any]
```

## `ConnectionAccount`

```python signature
class ConnectionAccount(id: str, label: str)
```

Whose account a connection is at the app.

## `ConnectionRejected`

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

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`

```python signature
class DefinitionError(*args, **kwargs)
```

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

## `FileReadAnswer`

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

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

## `FileStore`

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

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()`

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

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()`

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

`store`, from an async handler.

### `FileStore.read()`

```python signature
read(artifact_id: str) -> FileReadAnswer
```

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()`

```python signature
read_async(artifact_id: str) -> FileReadAnswer
```

`read`, from an async handler.

## `FileStoreAnswer`

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

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`

```python signature
class FileStoreError(*args, **kwargs)
```

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`

```python signature
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 = '',
)
```

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

## `Guardrail`

```python signature
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,
)
```

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

### `Guardrail.parse_config()`

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

### `Guardrail.parse_trace()`

```python signature
parse_trace(value: Any) -> Any
```

## `ModelCallRecord`

```python signature
class ModelCallRecord(**data: Any)
```

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`

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

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

## `PauseNotAvailable`

```python signature
class PauseNotAvailable(*args, **kwargs)
```

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`

```python signature
class RunTrace(**data: Any)
```

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`

```python signature
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,
)
```

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

### `Tool.parse_input()`

```python signature
parse_input(value: Any) -> Any
```

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

### `Tool.dump_output()`

```python signature
dump_output(value: Any) -> Any
```

The wire form of the handler's return value.

### `Tool.invoke_args()`

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

## `ToolCallRecord`

```python signature
class ToolCallRecord(**data: Any)
```

Fields:

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

## `ToolConnection`

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

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()`

```python signature
from_wire(value: Any) -> ToolConnection | None
```

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

## `ToolContext`

```python signature
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,
)
```

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()`

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

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()`

```python signature
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
```

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()`

```python signature
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
```

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`

```python signature
class ToolResultRecord(**data: Any)
```

Fields:

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

## `WaitToken`

```python signature
class WaitToken(id: str, url: str)
```

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

## `guardrail`

```python signature
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]
```

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`

```python signature
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]
```

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:

```plaintext
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`

```python signature
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]]
```

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.