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

CallCancelled

class CallCancelled(reason: CancelReason)

Raised by Cancellation.raise_if_cancelled once the call is over.

Cancellation

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

raise_if_cancelled() -> None

Cancellation.wait()

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

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

Cancellation.wait_async()

wait_async() -> CancelReason

Resolve once cancelled.

Cancellation.cancel()

cancel(reason: CancelReason) -> None

Fire (idempotent; the first reason wins).

CheckResult

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

to_wire() -> dict[str, Any]

DefinitionError

class DefinitionError(*args, **kwargs)

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

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

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

Guardrail.parse_config()

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

Guardrail.parse_trace()

parse_trace(value: Any) -> Any

ModelCallRecord

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

RunTrace

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)
• 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: 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,
    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()

parse_input(value: Any) -> Any

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

Tool.dump_output()

dump_output(value: Any) -> Any

The wire form of the handler's return value.

Tool.invoke_args()

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

ToolCallRecord

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

ToolContext

class ToolContext(
    tenant_id: str,
    run_id: str,
    request_id: str | None = None,
    env: Mapping[str, Any] = <factory>,
    secrets: Mapping[str, Any] = <factory>,
    config: Mapping[str, Any] = <factory>,
    cancellation: Cancellation = <factory>,
)

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

ToolContext.for_test()

for_test(
    tenant_id: str = 'tenant-test',
    run_id: str = 'run-test',
    **kwargs: Any,
) -> ToolContext

A context for calling a handler directly in a unit test.

ToolContext.from_wire()

from_wire(ctx: Mapping[str, Any], cancellation: Cancellation) -> ToolContext

The context for a protocol v2 ctx (tenantId, runId, …).

ToolResultRecord

class ToolResultRecord(**data: Any)

Fields:

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

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]

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}}). 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,
) -> 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:

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