# 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). ## `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, 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] ``` ## `DefinitionError` ```python signature class DefinitionError(*args, **kwargs) ``` A primitive is declared wrong — raised where it is declared. ## `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, 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` ## `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: 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()` ```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` ## `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] = , secrets: Mapping[str, Any] = , config: Mapping[str, Any] = , settings: Mapping[str, Mapping[str, Any]] = , cancellation: Cancellation = , log: Logger = Logger(subsystem='root', level='error'), idempotency_key: str | None = None, ) ``` Per-call context: who the call is for, its resolved env/secrets/config, its cancellation. ### `ToolContext.for_test()` ```python signature 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()` ```python signature from_wire( ctx: Mapping[str, Any], cancellation: Cancellation, log: Logger = Logger(subsystem='root', level='error'), ) -> ToolContext ``` The context for a protocol v2 `ctx` (`tenantId`, `runId`, …). ## `ToolResultRecord` ```python signature class ToolResultRecord(**data: Any) ``` Fields: - `tool_call_id: str` (`toolCallId` on the wire) - `output: Any` (optional) - `at: str` ## `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, ) -> 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` ```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, ) -> 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: 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.