loom.core.use_case

loom.core.use_case.Agent(name)[source]

Factory returning the runtime marker for a named agent handle parameter.

The executor resolves name against the agents compiled for this deployment and injects an AgentHandle bound to this execution’s verified caller — the only way a use case reaches an agent (constructor injection is not offered for this resource). The output type the handle carries is read from the parameter’s own AgentHandle[...] annotation, never from this factory, and is checked at start-up against the named agent’s declared output.

Returned value is intentionally typed as Any in overloads to avoid mypy default-argument incompatibility in signatures like: triage: AgentHandle[SeverityAssessment] = Agent("incident-triage").

Parameters:

name (str) – Name of a compiled agent, as declared by its artifact.

Return type:

Any

Example:

async def execute(
    self,
    caller: Identity = Caller(),
    triage: AgentHandle[SeverityAssessment] = Agent("incident-triage"),
) -> IncidentReport:
    assessment = await triage.run("Assess this incident.")
    ...
loom.core.use_case.Caller()[source]

Factory returning the runtime marker for the caller-identity parameter.

The executor injects the Identity the transport verified for this execution. It is a declaration, not an ambient read: the identity travels with the execution instead of hiding in a global.

Returned value is intentionally typed as Any in overloads to avoid mypy default-argument incompatibility in signatures like: caller: Identity = Caller().

Example:

async def execute(self, query: QuerySpec, caller: Identity = Caller()) -> Report:
    return await self._reports.for_owner(caller.require_subject(), query)
Return type:

Any

loom.core.use_case.Exists(entity_type, *, from_param=None, from_command=None, against, on_missing=OnMissing.RETURN_FALSE)[source]

Factory returning marker for boolean existence checks.

Parameters:

  • entity_type (type[EntityT])

  • from_param (str | None)

  • from_command (str | None)

  • against (str)

  • on_missing (OnMissing)

Return type:

Any

class loom.core.use_case.Compute[source]

Bases: object

Compute DSL namespace.

Example

Compute.set(F(UpdateUser).slug).from_command(F(UpdateUser).name, via=slugify)

loom.core.use_case.F(root)

Build a typed field-reference factory.

Example

F(UpdateUserCommand).birthdate

Parameters:

root (type[Command] | type[BaseModel] | str)

Return type:

Any

class loom.core.use_case.FieldRef(root, path)[source]

Bases: object

Declarative reference to a field path on a command (or loaded alias).

Parameters:
loom.core.use_case.Input()[source]

Factory returning the runtime marker for command payload parameters.

Returned value is intentionally typed as Any in overloads to avoid mypy default-argument incompatibility in signatures like: cmd: Command = Input().

Return type:

Any

loom.core.use_case.Load(entity_type, *, from_param=None, from_command=None, against, profile='default', on_missing=OnMissing.RAISE)[source]

Factory returning marker for preloaded entity parameters by field.

Parameters:

  • entity_type (type[EntityT])

  • from_param (str | None)

  • from_command (str | None)

  • against (str)

  • profile (str)

  • on_missing (OnMissing)

Return type:

Any

loom.core.use_case.LoadById(entity_type, *, by='id', profile='default', on_missing=OnMissing.RAISE)[source]

Factory returning marker for preloaded entity parameters by id.

Returned value is intentionally typed as Any in overloads to avoid mypy default-argument incompatibility in signatures like: entity: User = LoadById(User, by="id").

Parameters:

  • entity_type (type[EntityT]) – Domain entity type the repository should load.

  • by (str) – Name of the primitive parameter used as the lookup key. Defaults to "id".

  • profile (str) – Loading profile forwarded to repo.get_by_id. Defaults to "default".

  • on_missing (OnMissing) – Missing-entity policy. Defaults to OnMissing.RAISE.

Return type:

Any

loom.core.use_case.Mcp(server, *, include)[source]

Factory returning the runtime marker for a named MCP server handle parameter.

The executor resolves server against the MCP servers compiled for this deployment and injects an McpHandle bound to this execution’s verified caller — the only way a use case reaches an MCP server (constructor injection is not offered for this resource). Names in include are globs, matched by the same select_names/admits the model’s own toolset filter uses; there is no exclude in this version because no caller has asked for one and a short allow-list already expresses every case on the table. Under ai.remote_clients: optional, a server tolerated unreachable at start-up still resolves to a handle — every call on it fails with TOOL_UNAVAILABLE instead.

Unlike Agent(), no output type is ever checked against the parameter’s annotation: McpHandle carries no type parameter, so there is no declared shape to compare it with. Both checks a Mcp() marker gets are already wired at start-up, aborting the boot rather than waiting for a first call: server is validated against ai.mcp_servers, naming the declaring use case and parameter when it is not configured, and include is checked against the server’s real tool list under the same startup_timeout_ms an agent’s own mcp filter is checked against. The second check needs a listing, so under ai.remote_clients: optional a server that never connected is skipped rather than failing: a tolerated outage means the filter goes unverified, not that it verified clean.

Returned value is intentionally typed as Any to avoid mypy default-argument incompatibility in signatures like: search: McpHandle = Mcp("docs-server", include=["search"]).

Parameters:

  • server (str) – Name of a configured MCP server, as declared under ai.mcp_servers.

  • include (Sequence[str]) – Glob patterns naming the tools this handle may call. Keyword-only and required: everywhere this include/exclude shape is used, an empty include means “every name” — the filter only narrows when it carries at least one pattern — so an empty sequence here would silently grant the entire server, not the handful of tools the signature names. Mcp() raises ValueError instead of widening the grant behind the caller’s back. A bare str is rejected the same way: str satisfies Sequence[str], so include="search" would type-check yet split into six single-character glob patterns at runtime.

Raises:

ValueError – If include is empty, or is a single string instead of a sequence of patterns.

Return type:

Any

Example:

async def execute(
    self,
    caller: Identity = Caller(),
    docs: McpHandle = Mcp("docs-server", include=["search", "fetch"]),
) -> Report:
    names = docs.tools()
    ...
class loom.core.use_case.OnMissing(value)[source]

Bases: StrEnum

Policy applied when a marker lookup does not resolve an entity.

class loom.core.use_case.PredicateOp(value)[source]

Bases: StrEnum

class loom.core.use_case.Rule[source]

Bases: object

Rule DSL namespace.

class loom.core.use_case.RuleFn(*args, **kwargs)[source]

Bases: Protocol

class loom.core.use_case.UseCase(main_repo=None)[source]

Bases: ABC, Generic[ModelT, ResultT, RepoT]

Base class for all use cases.

Subclass and implement execute with typed parameters. Parameter defaults declare the execution contract:

  • Input() — command payload, built from the raw request.

  • LoadById(EntityType, by="param") — entity prefetched by id.

  • Load(EntityType, ...) — entity prefetched by arbitrary field.

  • Exists(EntityType, ...) — boolean existence check by field.

  • No default — primitive param bound directly from the caller.

Class attributes computes, rules, and read_only declare the pre-execution pipeline and execution policy. They are inspected once at startup by UseCaseCompiler and embedded in the immutable ExecutionPlan.

Parameters:

main_repo (RepoT | None)

computes

Compute transformations applied in order before rule checks.

Type:

ClassVar[Sequence[ComputeFn[Any]]]

rules

Rule validations applied in order after computes.

Type:

ClassVar[Sequence[RuleFn]]

read_only

When True, the executor skips opening a UnitOfWork transaction. Set this on query-only use cases that never mutate state. GET routes in RestInterface always bypass the UoW regardless of this flag.

Type:

ClassVar[bool]

Example:

class UpdateUserUseCase(UseCase[User, UserResponse]):
    computes = [set_updated_at]
    rules = [email_must_be_valid]

    def __init__(self, user_repo: UserRepository) -> None:
        self._user_repo = user_repo

    async def execute(
        self,
        user_id: int,
        cmd: UpdateUserCommand = Input(),
        user: User = LoadById(User, by="user_id"),
    ) -> UserResponse:
        ...
property main_repo: RepoT

Main repository injected by the factory.

classmethod describe_main_repo()[source]

Return the normalized main-repository contract for this use case.

Return type:

tuple[type[LoomStruct] | None, object]

abstractmethod async execute(*args, **kwargs)[source]

Execute core business logic.

Override with an explicit typed signature. The compiler inspects this method once at startup to build the ExecutionPlan.

Returns:

The result of the use case operation.

Parameters:
Return type:

ResultT