loom.ai.declarative

Authored agent artifacts: Tier-1 structs, envelope decoding and loading.

This package is the only supported entry point for artifacts a human or a generator writes. It is import-light on purpose: no optional extra is imported at module level, so import loom.ai.declarative works on a base install even though YAML artifacts need PyYAML.

Example

>>> specs = load_specs(["agents/*.agent.yaml"])
>>> [decoded.spec.name for decoded in specs]
['triage']

Functions

_decode_yaml_payload(buf, *, type)

_load_json(data, source)

_load_one(path)

_load_yaml(data, source)

_resolve_paths(patterns, root)

load_specs(patterns[, root])

Load every artifact matching a set of globs.

class loom.ai.declarative.A2ACapability(*, agent, include=(), exclude=())[source]

Bases: Struct

Delegation to a named remote agent reachable over A2A.

The artifact names the agent; ai.a2a_agents knows where it is and how to authenticate to it.

Parameters:
  • agent (Annotated[str, msgspec.Meta(min_length=1)]) – Named remote agent, resolved from ai.a2a_agents.

  • include (tuple[str, ...]) – Skill names or glob patterns to expose; empty means all.

  • exclude (tuple[str, ...]) – Skill names or glob patterns to omit, applied after include.

loom.ai.declarative.AgentSpec

alias of AgentSpecV1

class loom.ai.declarative.AgentSpecV1(*, spec_version, name, description, deps_type=None, deps_schema=None, instructions, model_role='default', output, output_check=None, on_output=None, conversation=None, capabilities=(), policies=<factory>, metadata=<factory>)[source]

Bases: Struct

Authored agent definition, format version 1.

Field order mirrors the published JSON Schema so an artifact reads the same way as the contract it validates against.

Parameters:
  • spec_version (Annotated[int, msgspec.Meta(ge=1, le=1)]) – Format version; always 1 for this struct.

  • name (Annotated[str, msgspec.Meta(pattern='^[a-z][a-z0-9_-]{0,62}$')]) – Unique agent name within the application.

  • description (Annotated[str, msgspec.Meta(min_length=1)]) – What the agent does. Published in the A2A card.

  • deps_type (Annotated[str, msgspec.Meta(pattern='^(dict|[A-Za-z_][A-Za-z0-9_.]*:[A-Za-z_][A-Za-z0-9_]*)$')] | None) – Declares the shape of the artifact’s state: the literal dict, or a module:Symbol reference matching DEPS_TYPE_PATTERN. Sugar over deps_schema (FR-003). dict contains no colon, so the two forms cannot collide. None when the artifact declares no state.

  • deps_schema (Mapping[str, Any] | None) – Declares the shape of the artifact’s state directly, as a JSON Schema object — the canonical form of the one mechanism deps_type is sugar over (FR-003). None when the artifact declares no state, or declares it through deps_type.

  • instructions (Annotated[str, msgspec.Meta(min_length=1)] | Annotated[tuple[InstructionBlock, ...], msgspec.Meta(min_length=1)]) – Instructions the agent follows: a literal string, or a non-empty sequence of InstructionBlock in authored order. Never published, and never a place to encode authorization.

  • model_role (Annotated[str, msgspec.Meta(pattern='^[a-z][a-z0-9_-]{0,31}$')]) – Logical model role bound to a concrete provider and model by deployment configuration.

  • output (JsonSchemaOutput | TypeRefOutput) – Declaration of the structured answer the agent returns.

  • output_check (Annotated[str, msgspec.Meta(pattern='^[A-Za-z_][A-Za-z0-9_.]*:[A-Za-z_][A-Za-z0-9_]*$')] | None) – module:symbol reference to an OutputCheck, matching SYMBOL_REF_PATTERN. Resolved at compile time; None when the artifact declares no check.

  • on_output (OutputHookSpec | None) – Use case executed once per completed run with the validated output; None when the artifact declares no hook.

  • conversation (ConversationSpec | None) – Use case executed before a run that carries a conversation_id to load the prior history; None when the artifact declares no loader.

  • capabilities (tuple[UsecaseCapability | SqlCapability | McpCapability | SkillsCapability | PythonCapability | A2ACapability | NativeCapability, ...]) – Explicitly granted capabilities; empty by default.

  • policies (PolicySpec) – Execution limits; documented defaults when omitted.

  • metadata (Mapping[str, str]) – Free-form string labels carried alongside the agent.

class loom.ai.declarative.ConversationSpec(*, usecase)[source]

Bases: Struct

Use case the runtime executes before a run that carries a conversation_id.

It returns the prior history of that conversation as opaque bytes in the engine’s serialised form, or None on the first turn. The key uses the same vocabulary as UsecaseCapability.keys and is resolved against the same registry at compile time. The model never sees it: it is not a tool, and it never enters the instructions.

Parameters:

usecase (Annotated[str, msgspec.Meta(min_length=1)]) – Use-case key of the registry that loads the prior history.

class loom.ai.declarative.DecodedSpec(*, spec, issues=(), source_path=None)[source]

Bases: Struct

One successfully decoded artifact and its non-fatal findings.

Parameters:
  • spec (AgentSpecV1) – The decoded artifact.

  • issues (tuple[AgentCompilationIssue, ...]) – Non-fatal issues raised while decoding, such as a deprecation notice for a superseded but still readable spec version.

  • source_path (str | None) – File the artifact was read from, when there is one. A ./ skill library resolves against this path, so an artifact decoded from bare bytes cannot use one.

class loom.ai.declarative.InstructionBlock(*, text, name=None, template=None)[source]

Bases: Struct

One authored instruction block.

A bare string instructions is sugar for a single unnamed block with no template; a sequence of blocks is authored order, projected onto the engine in that same order.

dynamic is not authored here: it is not cosmetic, it decides what a provider may cache, and it follows from whether template is declared — an author who could set it independently could only get it wrong (FR-025).

Parameters:
  • text (Annotated[str, msgspec.Meta(min_length=1)]) – Instruction text. Literal unless template names a template engine; with no template, any {{ it contains reaches the model unchanged (FR-022).

  • name (Annotated[str, msgspec.Meta(pattern='^(?!agent$)[^:]+$')] | None) – Optional name identifying the block in compilation issues and start-up diagnostics; matches INSTRUCTION_NAME_PATTERN. It never becomes an addressable id on the engine’s own side (FR-026).

  • template (Literal['handlebars'] | None) – Names the template engine text is written for, one of TEMPLATE_ENGINES. None when text is a literal string (FR-022).

class loom.ai.declarative.JsonSchemaOutput(*, schema)[source]

Bases: Struct

Structured answer described by an inline JSON Schema object.

Canonical output form: what a generator emits.

Parameters:

schema (Mapping[str, Any]) – JSON Schema object describing the required answer.

class loom.ai.declarative.McpCapability(*, server, include=(), exclude=())[source]

Bases: Struct

Tools served by a named remote MCP server.

The artifact names the server; it never locates it. Where the server lives, how to authenticate to it and how long to wait are deployment facts read from ai.mcp_servers, so the same artifact moves between environments unchanged.

Parameters:
  • server (Annotated[str, msgspec.Meta(min_length=1)]) – Named server, resolved from ai.mcp_servers.

  • include (tuple[str, ...]) – Tool names or glob patterns to expose; empty means all.

  • exclude (tuple[str, ...]) – Tool names or glob patterns to omit, applied after include.

class loom.ai.declarative.NativeCapability(*, tool)[source]

Bases: Struct

Tool the model provider executes in its own infrastructure.

The artifact names the tool; whether the model bound to model_role admits it is a deployment fact checked at compile time, and the provider runs it, so no toolset, timeout or credential of loom is involved.

Parameters:

tool (Literal['web_search', 'web_fetch', 'code_execution']) – Provider tool, one of NATIVE_TOOLS.

class loom.ai.declarative.OutputHookSpec(*, usecase)[source]

Bases: Struct

Use case the runtime executes once per completed run, with the validated output.

The key uses the same vocabulary as UsecaseCapability.keys and is resolved against the same registry at compile time. The model never sees it: it is not a tool, and it never enters the instructions.

Parameters:

usecase (Annotated[str, msgspec.Meta(min_length=1)]) – Use-case key of the registry to execute with the validated output.

class loom.ai.declarative.PolicySpec(*, retries=2, tool_timeout_ms=20000, max_iterations=12, run_timeout_ms=120000, max_history_bytes=1048576, max_usd=None, max_total_tokens=None, max_input_tokens_per_request=None, max_tool_calls=None, max_requests=50, on_unpriced_spend='serve')[source]

Bases: Struct

Execution limits an agent runs under.

Ranges are published as module constants and enforced by a later compilation phase, so an out-of-range value is reported as a coded issue rather than as a decoding failure. See “Spend caps” and “max_iterations versus max_tool_calls” in docs/ai/artifacts.md for the rationale behind the fields below.

Parameters:
  • retries (int) – Retries a failed tool call, and an answer output_check rejects, inside one run, always. Retries a failed provider call across runs, only when the plan holds no capability. See RETRIES_DESCRIPTION and “retries” in docs/ai/artifacts.md.

  • tool_timeout_ms (int) – Deadline of a single tool call.

  • max_iterations (int) – Maximum ToolCallEvents loom’s own supervisor observes over the event stream in one run; see “max_iterations versus max_tool_calls” in docs/ai/artifacts.md for how it differs from max_tool_calls.

  • run_timeout_ms (int) – Deadline of a whole run.

  • max_history_bytes (int) – Ceiling, in bytes, of the history a conversation loader may return; a longer one fails the run.

  • max_usd (Decimal | None) – Cumulative spend ceiling in US dollars for one run, including every retried attempt — not for a conversation. None disables the cap. Projects onto UsageLimits.cost_limit; see “Spend caps” in docs/ai/artifacts.md for enforcement timing and the YAML/JSON precision difference.

  • max_total_tokens (int | None) – Cumulative input-plus-output token ceiling for the whole run. None disables the cap. Projects onto UsageLimits.total_tokens_limit; see “Spend caps” in docs/ai/artifacts.md.

  • max_input_tokens_per_request (int | None) – Ceiling on the input tokens of any one request in the run. None disables the cap. Projects onto UsageLimits.per_request_input_tokens_limit; see “Spend caps” in docs/ai/artifacts.md.

  • max_tool_calls (int | None) – Cumulative successful tool-call ceiling for the whole run. None disables the cap. Projects onto UsageLimits.tool_calls_limit; see “Spend caps” in docs/ai/artifacts.md.

  • max_requests (int) – Cumulative model-request ceiling for the whole run, counted by the engine. Defaults to MAX_REQUESTS_DEFAULT. Projects onto UsageLimits.request_limit; see “Spend caps” in docs/ai/artifacts.md.

  • on_unpriced_spend (Literal['serve', 'refuse']) – What a run does when max_usd is declared and at least one of its model responses could not be priced. Inert when max_usd is absent; see “Spend caps” in docs/ai/artifacts.md.

class loom.ai.declarative.PythonCapability(*, factory, params=<factory>)[source]

Bases: Struct

Toolset built by application-owned Python code.

Parameters:
  • factory (Annotated[str, msgspec.Meta(pattern='^[A-Za-z_][A-Za-z0-9_.]*:[A-Za-z_][A-Za-z0-9_]*$')]) – module:factory called once at build as factory(context, **params). A factory, never a constructed object.

  • params (dict[str, Any]) – Nested block passed to the factory as keyword arguments. The names are validated against the factory’s signature at compile; the values are decoded YAML, not validated. Settings, never secrets.

class loom.ai.declarative.SkillsCapability(*, library, include=(), exclude=())[source]

Bases: Struct

Packaged prompt material from one skill library.

The artifact names a library; it never carries an absolute path. ./name resolves beside the artifact and travels with it, a bare name resolves against ai.skills_root, and .. is not representable, so a library can never escape its own directory.

Parameters:
  • library (Annotated[str, msgspec.Meta(pattern='^(\\./[A-Za-z0-9._-]+|[A-Za-z0-9._-]+)$', min_length=1)]) – Skill library, either ./name or a bare name.

  • include (tuple[str, ...]) – Skill names or glob patterns to expose; empty means all.

  • exclude (tuple[str, ...]) – Skill names or glob patterns to omit, applied after include.

class loom.ai.declarative.SqlCapability(*, connection, max_rows, max_result_bytes)[source]

Bases: Struct

Read-only access to a named SQL connection.

Result bounds are mandatory: an unbounded query is not representable (FR-046b).

Parameters:
  • connection (Annotated[str, msgspec.Meta(min_length=1)]) – Named connection; compilation fails unless it is read-only.

  • max_rows (Annotated[int, msgspec.Meta(ge=1)]) – Maximum number of rows a single query may return.

  • max_result_bytes (Annotated[int, msgspec.Meta(ge=1)]) – Maximum size of a single query result.

class loom.ai.declarative.TypeRefOutput(*, ref)[source]

Bases: Struct

Structured answer described by an application type.

Shortcut for hand-written applications; the reference is resolved at compile time.

Parameters:

ref (Annotated[str, msgspec.Meta(pattern='^[A-Za-z_][A-Za-z0-9_.]*:[A-Za-z_][A-Za-z0-9_]*$')]) – module:Symbol reference to the answer type.

class loom.ai.declarative.UsecaseCapability(*, keys)[source]

Bases: Struct

Explicitly granted business operations.

Parameters:

keys (Annotated[tuple[Annotated[str, msgspec.Meta(min_length=1)], ...], msgspec.Meta(min_length=1)]) – Use-case keys granted to the agent. Never expanded automatically.

loom.ai.declarative.agent_spec_json_schema(spec_version=LATEST_SPEC_VERSION)[source]

Return the published JSON Schema document for a spec version.

The document is rebuilt on every call, so callers may mutate the result freely without affecting anyone else.

Parameters:

spec_version (int) – Spec version whose schema is wanted.

Returns:

The JSON Schema document, ready to serialise or hand to a validator.

Raises:

ValueError – If no schema is published for that version.

Return type:

dict[str, Any]

Example

>>> agent_spec_json_schema(1)["title"]
'Loom AgentSpec v1'
loom.ai.declarative.agent_spec_schema_path(spec_version=LATEST_SPEC_VERSION)[source]

Locate the JSON Schema file shipped in the installed distribution.

The file is the byte-for-byte twin of agent_spec_json_schema(), kept honest by a test. Prefer this over the emitter when the schema must be handed to a tool that reads a path — an editor, a linter, a CI validation step in another language.

Parameters:

spec_version (int) – Spec version whose schema file is wanted.

Returns:

Absolute path of the shipped schema file.

Raises:
  • ValueError – If no schema is published for that version.

  • FileNotFoundError – If the distribution was built without its schema data files, which is a packaging defect rather than a usage error.

Return type:

Path

Example

>>> agent_spec_schema_path(1).name
'agent-spec-v1.schema.json'
loom.ai.declarative.decode_spec(data, *, source=ANONYMOUS_SOURCE, versions=SUPPORTED_SPEC_VERSIONS)[source]

Decode JSON artifact bytes into the struct of the version they declare.

Parameters:
  • data (bytes) – Raw JSON bytes of a single artifact.

  • source (str) – Human-readable origin used as every issue’s component and, when it is a real path, as the artifact’s source_path.

  • versions (Mapping[int, type[AgentSpecV1]]) – Registry of readable spec versions. Overriding it is how the deprecation path is exercised without shipping a fictitious future version: a registry whose maximum key is above the artifact’s version turns that artifact into a deprecated-but-readable one.

Returns:

The decoded artifact together with its non-fatal issues, such as a deprecation notice.

Raises:

AgentCompilationError – If spec_version is absent, unsupported, or the payload does not decode as the declared version.

Return type:

DecodedSpec

Example

>>> decoded = decode_spec(raw_bytes, source="agents/triage.agent.yaml")
>>> decoded.spec.name
'triage'
loom.ai.declarative.load_specs(patterns, root='.')[source]

Load every artifact matching a set of globs.

Patterns are resolved relative to root; matches are de-duplicated across patterns and returned sorted by path, so the result is deterministic. .yaml/.yml files decode as YAML, .json files as JSON, and any other extension is a failure.

Failures from every file are accumulated and reported once, so a broken artifact does not hide the ones after it. Non-fatal findings ride on each returned DecodedSpec. Duplicate agent names are not checked here: that is a later compilation phase over the whole application.

Parameters:
  • patterns (Sequence[str]) – Glob patterns, relative to root.

  • root (Path | str) – Directory the patterns are resolved against.

Returns:

One decoded artifact per matched file, ordered by path.

Raises:

AgentCompilationError – Aggregating every fatal issue found across all matched files.

Return type:

tuple[DecodedSpec, …]

Example

>>> load_specs(["*.agent.yaml"], root="agents")
(DecodedSpec(spec=AgentSpecV1(...), issues=()),)