Skip to content

angee.agents.models

Source models for the agent catalogue.

An :class:Agent is a definition the operator later renders into a workspace and service. It draws on three catalogues this addon also owns: :class:Skill rows discovered from an integrate.Source, :class:MCPServer/:class:MCPTool rows, and an :class:InferenceProvider integration child with its :class:InferenceModel rows. Templates are agents with is_template set. This addon keeps definitions only; the operator owns lifecycle.

InferenceModelUse

python
class InferenceModelUse(models.TextChoices)

What an inference model is used for (mirrors the LLM catalogue's model use).

InferenceModelStatus

python
class InferenceModelStatus(models.TextChoices)

Lifecycle of a model in a provider's catalogue.

MCPPlacement

python
class MCPPlacement(models.TextChoices)

Where an MCP server runs relative to the platform.

MCPTransport

python
class MCPTransport(models.TextChoices)

How an agent reaches an MCP server.

BUILTIN_MCP_ANGEE

MCPServer.config["builtin"] value for this process's built-in Angee MCP server.

ToolGrant

Table-less REBAC type anchor for pure-tuple tool grants.

ToolRole

Table-less REBAC type anchor for hierarchical tool bundles.

AgentLifecycle

python
class AgentLifecycle(models.TextChoices)

Where an agent sits in the operator provision pipeline.

The lifecycle axis: a forward journey the render pipeline drives, distinct from the agent's observed run state (:class:RuntimeStatus). A provision moves it DRAFT → PROVISIONING → READY; a teardown moves it → DEPROVISIONING → DEPROVISIONED. Whether the rendered agent is actually up — and whether the last operation failed — is the orthogonal :attr:Agent.runtime_status, never folded in here.

RuntimeStatus

python
class RuntimeStatus(models.TextChoices)

The observed run state of a provisioned thing — the colored-dot axis.

Orthogonal to a provision lifecycle (:class:AgentLifecycle): stopped/running/ error/warning is "is it up right now, and is anything wrong", the grey/green/red/ amber dot the frontend renders through the colorDot widget. Reused for any model that has a run state; the operator daemon reports the same vocabulary for its services (see docs/frontend/guidelines.md for the shared tone mapping).

SessionStatus

python
class SessionStatus(models.TextChoices)

Display projection of one persisted agent session's workflow state.

TurnStatus

python
class TurnStatus(models.TextChoices)

Execution state of one prompt-to-response turn.

InferenceProvider

python
class InferenceProvider(ImplDefaultsMixin, AngeeModel)

An LLM provider account, materialized as an integration child row.

It draws its API credential from its inherited integration credential and resolves its provider-specific :class:~angee.agents.backends.InferenceBackend from backend_class. Django keeps the catalogue; the backend lists the provider's models into :class:InferenceModel rows.

backend_class

Registry key for the inference backend this provider uses.

base_url

Base endpoint for OpenAI-compatible providers; blank uses the backend default.

config

Provider-scoped settings used by inference implementations.

Meta

python
class Meta()

Django model options for the inference provider child model.

__str__

python
def __str__() -> str

Return the provider's display label.

backend

python
@property
def backend() -> InferenceBackend

Return the backend bound to this provider's credential and endpoint.

Resolved fresh per access (unlike storage.Backend.storage, which caches): the built-in manual backend holds no client, and a vendor backend reads the live credential off this provider each call, so a rotated key takes effect at once and the backend owns any client lifetime it needs.

refresh_models

python
def refresh_models() -> int

Re-list this provider's models into :class:InferenceModel rows.

chat

python
def chat(*,
         model: str,
         messages: Sequence[Mapping[str, Any]],
         system: str = "",
         max_tokens: int = 1024,
         temperature: float | None = None,
         tools: Sequence[Mapping[str, Any]] = (),
         options: Mapping[str, Any] | None = None) -> InferenceResponse

Send one non-streaming chat request through this provider.

InferenceModelManager

python
class InferenceModelManager(AngeeManager)

Manager owning the upsert of model rows from a provider's catalogue.

sync_from_provider

python
def sync_from_provider(provider: Any) -> int

Upsert one row per model the provider advertises (non-destructive).

Missing handles are left in place, not pruned, so an agent's model FK is never broken by a transient provider response; deprecation is a status edit.

InferenceModel

python
class InferenceModel(SqidMixin, AuditMixin, AngeeModel)

One model in a provider's catalogue, agents bind to by FK.

publisher is the model's maker, reusing the integrate.Vendor catalogue (which need not be the serving provider's vendor — an OpenAI-compatible router can serve another maker's model).

name

The selectable runtime handle; config.provider_model carries the native provider id.

Meta

python
class Meta()

Django model options for catalogue models.

__str__

python
def __str__() -> str

Return the model's display label.

provider_model_name

python
@property
def provider_model_name() -> str

Return the provider-native model id for runtimes that talk to the provider directly.

credential

python
@property
def credential() -> Any

Return the API credential for this model, via its provider's integration.

chat

python
def chat(messages: Sequence[Mapping[str, Any]],
         *,
         system: str = "",
         max_tokens: int = 1024,
         temperature: float | None = None,
         tools: Sequence[Mapping[str, Any]] = (),
         options: Mapping[str, Any] | None = None) -> InferenceResponse

Send one non-streaming chat request through this catalogue model.

SkillManager

python
class SkillManager(AngeeManager)

Manager owning the reconcile of skill rows from a skill source.

sync_from_source

python
def sync_from_source(source: Any) -> int

Walk the source for SKILL.md and upsert/prune :class:Skill rows.

Skill

python
class Skill(SqidMixin, AuditMixin, AngeeModel)

One skill discovered under an integrate.Source (source_kind="skill").

The operator mounts the skill's directory into an agent's workspace; Django keeps the inventory only. Discovery reuses the integrate source walk.

source_kind

Binds the skill source kind to this output model (see integrate.Source).

Meta

python
class Meta()

Django model options for discovered skills.

__str__

python
def __str__() -> str

Return the skill's display label.

MCPServer

python
class MCPServer(SqidMixin, AuditMixin, AngeeModel)

An MCP server an agent can reach — internal to the platform or external.

An external server authenticates with an integrate.Credential; the operator renders the selected servers and their authorized tools into an agent's MCP config.

Meta

python
class Meta()

Django model options for MCP servers.

__str__

python
def __str__() -> str

Return the server's name.

builtin

python
@property
def builtin() -> str

Return the built-in MCP server key this row targets, if any.

resolved_url

python
@property
def resolved_url() -> str

Return the container-reachable URL this MCP server renders for agents.

Explicit url wins for external/custom servers. The built-in Angee MCP server is modelled as config = {"builtin": "angee"} with no row-owned URL; the stack supplies the concrete container-reachable URL through ANGEE_BUILTIN_MCP_URL so demo/catalogue data never bakes in a dev port.

is_addressable

python
@property
def is_addressable() -> bool

Whether a rendered container can reach this server — i.e. it has a URL.

A stdio server is a local command with no URL and isn't rendered into an agent's .mcp.json. A built-in Angee server is addressable when the stack supplied ANGEE_BUILTIN_MCP_URL.

config_entry

python
def config_entry(bearer_env: str | None) -> dict[str, Any]

Return this server's .mcp.json entry, given an optional bearer env var.

A credentialed server carries an Authorization: Bearer header whose value is the agent runtime's ${<bearer_env>} expansion — the bearer rides the container env (set from the operator secret in the service env, like the inference token), never the file or browser. The operator only resolves ${secret.<name>} in a service's env, not in file content, so a bearer placed literally in .mcp.json would never resolve. Pass None for an uncredentialed server.

MCPTool

python
class MCPTool(SqidMixin, AuditMixin, AngeeModel)

One tool an MCP server exposes; agents select the tools they may call.

requires_approval

Whether invoking this tool must suspend for an owner decision.

Meta

python
class Meta()

Django model options for MCP tools.

__str__

python
def __str__() -> str

Return the tool's name.

AgentManager

python
class AgentManager(AngeeManager)

Manager owning service-user lifecycle for agent principals.

service_username

python
def service_username(agent: Any) -> str

Return the deterministic username for agent's service user.

sync_service_user

python
def sync_service_user(agent: Any) -> Any

Create or update agent's non-login service user.

The service row is system-owned attribution state, not actor-authored profile data, so it is written elevated and keyed only by the agent's stable sqid-derived username.

deactivate_service_user

python
def deactivate_service_user(agent: Any, *, using: str | None = None) -> None

Deactivate agent's linked service user, leaving attribution FKs intact.

Agent

python
class Agent(SqidMixin, AuditMixin, AngeeModel)

An agent definition (or, when is_template, an agent template).

The operator renders an agent into a workspace from workspace_template and a service from the template its runtime_class declares, writing the instructions into AGENTS.md/CLAUDE.md, the selected skills and MCP servers/tools into the workspace, and the model's API credential into the service. service and workspace hold the operator instance names once rendered.

user

Non-login service user used for audit/revision attribution by this agent.

instructions

The agent's system instructions, rendered into AGENTS.md/CLAUDE.md.

inference_credential

Per-agent inference credential override. When set, the agent authenticates inference with this credential (e.g. a connected Anthropic OAuth account) instead of the one its model's provider integration carries; unset falls back to that catalogue chain.

runtime_class

Registry key for the agent runtime — the program this agent renders into. The runtime owns its operator service template, how it consumes an inference credential as container env, and the model-handle convention; none renders no service (a workspace-only agent).

service

Operator service instance name, set when the agent is rendered.

workspace

Operator workspace instance name, set when the agent is rendered.

lifecycle

Provision-pipeline position (:class:AgentLifecycle), set by the render flow.

runtime_status

Observed run state (:class:RuntimeStatus) — the colored dot; ERROR pairs with last_error. Set by the render flow; the daemon owns the live truth.

last_error

The reason runtime_status is ERROR — the last failed operation.

Meta

python
class Meta()

Django model options for agents.

__str__

python
def __str__() -> str

Return the agent's name.

save

python
def save(*args: Any, **kwargs: Any) -> None

Persist the agent and sync its service-user label.

Mirrors iam.User.save(): the row save owns a small derived sync, and the manager performs the system-owned dependent write.

principal_subject

python
def principal_subject() -> SubjectRef

Return this agent's own REBAC subject identity for actions it performs.

This is distinct from :attr:owner: the owner manages the agent definition, while the agent subject represents the running agent as an actor.

runtime_backend

python
@property
def runtime_backend() -> AgentRuntime

Return the :class:~angee.agents.runtimes.AgentRuntime this agent renders into.

Resolved fresh from runtime_class per access (the runtime is stateless); it owns the service template, the credential→env mapping, and the model handle.

can_provision

python
@property
def can_provision() -> bool

Whether the provision action may start from the current lifecycle facts.

can_deprovision

python
@property
def can_deprovision() -> bool

Whether the teardown action is meaningful for the current rendered state.

can_delete

python
@property
def can_delete() -> bool

Whether deleting the definition can leave no orphaned operator instance.

delete_blocker

python
def delete_blocker() -> str | None

Return the delete-blocking reason, or None when deletion is allowed.

mark_provisioning

python
@transition(
    lifecycle,
    source=[
        AgentLifecycle.DRAFT,
        AgentLifecycle.PROVISIONING,
        AgentLifecycle.READY,
        AgentLifecycle.DEPROVISIONED,
    ],
    target=AgentLifecycle.PROVISIONING,
    on_success=save_state,
)
def mark_provisioning() -> None

Enter the provision flow: lifecycle provisioning, run state reset to stopped.

mark_workspace_provisioned

python
@transition(
    lifecycle,
    source=AgentLifecycle.PROVISIONING,
    target=AgentLifecycle.PROVISIONING,
    on_success=save_state,
)
def mark_workspace_provisioned(*, workspace: str) -> None

Record the workspace as soon as the operator creates it.

mark_service_provisioned

python
@transition(
    lifecycle,
    source=AgentLifecycle.PROVISIONING,
    target=AgentLifecycle.PROVISIONING,
    on_success=save_state,
)
def mark_service_provisioned(*, service: str) -> None

Record the service as soon as the operator creates it.

mark_provisioned

python
@transition(
    lifecycle,
    source=AgentLifecycle.PROVISIONING,
    target=AgentLifecycle.READY,
    on_success=save_state,
)
def mark_provisioned(*, workspace: str, service: str = "") -> None

Record the operator instance the provision flow rendered for this agent.

The daemon owns the workspace/service lifecycle; the server-side provision flow renders them and calls this to persist the resulting instance names, mark the lifecycle READY and the run state RUNNING. Clears any prior error. service is optional — a workspace-only agent renders no service.

mark_deprovisioned

python
@transition(
    lifecycle,
    source=[
        AgentLifecycle.DRAFT, AgentLifecycle.DEPROVISIONING,
        AgentLifecycle.DEPROVISIONED
    ],
    target=AgentLifecycle.DEPROVISIONED,
    on_success=save_state,
)
def mark_deprovisioned() -> None

Clear the operator instance after teardown: lifecycle deprovisioned, run state stopped.

mark_deprovisioning

python
@transition(
    lifecycle,
    source=[
        AgentLifecycle.DRAFT,
        AgentLifecycle.PROVISIONING,
        AgentLifecycle.READY,
        AgentLifecycle.DEPROVISIONING,
        AgentLifecycle.DEPROVISIONED,
    ],
    target=AgentLifecycle.DEPROVISIONING,
    on_success=save_state,
)
def mark_deprovisioning() -> None

Mark the agent as tearing down through the operator teardown flow.

mark_provision_failed

python
def mark_provision_failed(message: str,
                          *,
                          clear_instances: bool = False,
                          clear_service: bool = False) -> None

Record a failed operation: run state ERROR (the red dot), reason kept.

The failure lands on the run-state axis: last_error holds the reason and the dot turns red. clear_instances blanks both instance names (a provision rolled the workspace back); clear_service blanks only the service (a reprovision destroyed the old service before the recreate failed — the workspace is preserved). The lifecycle then follows the workspace, never stranding mid-flow: an agent left holding a workspace is still provisioned (READY), one rolled back to nothing is a clean DRAFT retry. The red run-state dot carries the failure either way, and the persisted names never point at a torn-down instance. This deliberately bypasses the declared lifecycle graph because the target is data-dependent recovery state, not a user-visible lifecycle action.

provision_workspace_inputs

python
def provision_workspace_inputs() -> dict[str, str]

Resolve the agent-default workspace template inputs from this agent.

The structured fields (name, instructions, MCP servers) are the source of truth, so they win over any same-named key in workspace_inputs (which carries template-specific extras only). All values are stringified — Copier and the daemon take string answers.

provision_service_inputs

python
def provision_service_inputs() -> dict[str, str]

Resolve the structured service-template inputs from this agent.

Carries the runtime model handle plus the runtime-owned auth env block. The secret value never appears here — only operator ${secret.…} placeholders; the value is synced server-side. The runtime (not the provider) owns how the credential becomes env, because the same token feeds different env vars in different runtimes. service_inputs supplies template-specific extras (permission_mode etc.) and loses to the structured keys.

service_model_handle

python
def service_model_handle() -> str

Return the selected model handle in this agent's runtime convention.

mcp_config

python
def mcp_config() -> dict[str, Any]

Return the .mcp.json document for this agent's reachable MCP servers.

Each server renders its own entry (:meth:MCPServer.config_entry); this supplies the bearer env var (:meth:mcp_bearer_env) for a credentialed server and skips servers that aren't addressable. The header expands that env var, which the service env sets from the operator secret (:meth:provision_service_inputs) — the value rides the container env, never the file or the browser. See :meth:mcp_secrets.

mcp_secret_name

python
def mcp_secret_name(server: MCPServer) -> str

Return the operator secret name holding one MCP server's bearer for this agent.

Stable and scoped to the agent + server credential: the service env references it (${secret.<name>} → the bearer env var) and the provision flow syncs the credential value under it (the value never appears in the file or the browser).

mcp_bearer_env

python
def mcp_bearer_env(server: MCPServer) -> str

Return the container env var carrying one MCP server's bearer for this agent.

The rendered .mcp.json header reads it via the agent runtime's ${VAR} expansion; the service env sets it from the operator secret (:meth:mcp_secret_name), which the operator resolves into the container env. Keyed by the server credential so the env name and the .mcp.json reference can't drift, and unique per server in the agent's container. The sqid segment is upper-cased so the env name is portable across container runtimes/shells that reject or fold lowercase env names.

mcp_secrets

python
def mcp_secrets() -> dict[str, str]

Return {secret_name: bearer_value} for every credentialed MCP server.

Server-side only — the provision flow pushes these to the operator secret store so each server's ${secret.<name>} header resolves in the container.

inference_secret_name

python
def inference_secret_name() -> str

Return the operator secret name holding this agent's inference token.

Stable and agent-scoped — the provision inputs reference it and the (server-side) secret sync writes the credential value under it.

inference_secret

python
def inference_secret() -> str

Return the inference credential's secret value (API key or OAuth token), or "".

Server-side only — the value is pushed to the operator secret store under inference_secret_name() and never returned to the browser. An OAuth token near expiry is renewed first (:meth:integrate.Credential.ensure_fresh) so the value frozen into the provisioned service has its full lifetime ahead of it.

provision_inference_secret

python
def provision_inference_secret() -> str

Return the runtime-shaped inference secret payload synced to the operator store.

The value the provision flow stores under :meth:inference_secret_name, that the service's ${secret.<name>} auth placeholder resolves to in the container: the raw credential secret for most runtimes, or a runtime-built payload (OpenCode's base64 auth.json for an OAuth credential — see :meth:~angee.agents.runtimes.AgentRuntime.auth_secret_value). "" when there is no credential to sync. Readiness still gates on :meth:inference_secret (the raw token), so an empty credential is refused before this richer payload is built. A runtime that renders no service has no container to consume the secret, so nothing is synced (kept in step with :meth:provision_service_inputs' auth-env block).

inference_credential_ready

python
def inference_credential_ready() -> bool

Whether this agent can be provisioned with working inference auth.

A model-less agent needs no inference credential, so it is always ready. A credential-less backend is ready only for an in-process runtime, where its default endpoint reaches the host directly. Every attached credential must yield a usable secret, and a service-rendering runtime must be able to consume its kind, so an unworkable pairing is refused here rather than degrading silently at run time.

inference_credential_for_runtime

python
def inference_credential_for_runtime() -> Any

Return the integrate.Credential backing this agent's inference, or None.

A per-agent inference_credential override wins (e.g. a connected Anthropic OAuth account the user pointed this agent at); otherwise the model's catalogue credential (the model→provider→credential chain the catalogue owns), asked of the model rather than walked here.

AgentSession

python
class AgentSession(SqidMixin, AuditMixin, AngeeModel)

Runtime-neutral persisted conversation backed by one workflow run.

Meta

python
class Meta()

Django model options for persisted agent sessions.

__str__

python
def __str__() -> str

Return the session title or its agent name.

mark_running

python
@transition(
    status,
    source=[
        SessionStatus.IDLE, SessionStatus.RUNNING,
        SessionStatus.AWAITING_APPROVAL, SessionStatus.ERROR
    ],
    target=SessionStatus.RUNNING,
    on_success=save_state,
)
def mark_running() -> None

Project active turn execution onto the session.

mark_idle

python
@transition(
    status,
    source=[SessionStatus.IDLE, SessionStatus.RUNNING, SessionStatus.ERROR],
    target=SessionStatus.IDLE,
    on_success=save_state,
)
def mark_idle() -> None

Project a parked session awaiting its next user turn.

mark_awaiting_approval

python
@transition(
    status,
    source=SessionStatus.RUNNING,
    target=SessionStatus.AWAITING_APPROVAL,
    on_success=save_state,
)
def mark_awaiting_approval() -> None

Project a suspended tool approval onto the session.

close

python
@transition(
    status,
    source=[
        SessionStatus.IDLE, SessionStatus.RUNNING,
        SessionStatus.AWAITING_APPROVAL, SessionStatus.ERROR
    ],
    target=SessionStatus.CLOSED,
    on_success=save_state,
)
def close() -> None

Close the conversation so its workflow step can finish.

mark_error

python
@transition(
    status,
    source=[
        SessionStatus.IDLE, SessionStatus.RUNNING,
        SessionStatus.AWAITING_APPROVAL, SessionStatus.ERROR
    ],
    target=SessionStatus.ERROR,
    on_success=save_state,
)
def mark_error(message: str) -> None

Project a session-level runtime error.

AgentTurn

python
class AgentTurn(SqidMixin, AuditMixin, AngeeModel)

One prompt-to-response cycle with append-only ACP updates.

Meta

python
class Meta()

Django model options for agent turns.

mark_running

python
@transition(
    status,
    source=[
        TurnStatus.PENDING, TurnStatus.RUNNING, TurnStatus.AWAITING_APPROVAL
    ],
    target=TurnStatus.RUNNING,
    on_success=save_state,
)
def mark_running() -> None

Claim or resume this turn for runtime execution.

mark_awaiting_approval

python
@transition(
    status,
    source=TurnStatus.RUNNING,
    target=TurnStatus.AWAITING_APPROVAL,
    on_success=save_state,
)
def mark_awaiting_approval() -> None

Suspend this turn for deferred tool approval.

mark_completed

python
@transition(
    status,
    source=[TurnStatus.RUNNING, TurnStatus.AWAITING_APPROVAL],
    target=TurnStatus.COMPLETED,
    on_success=save_state,
)
def mark_completed(*, text: str, usage: Mapping[str, int]) -> None

Persist the final response projection for a completed turn.

mark_failed

python
@transition(
    status,
    source=[TurnStatus.RUNNING, TurnStatus.AWAITING_APPROVAL],
    target=TurnStatus.FAILED,
    on_success=save_state,
)
def mark_failed(message: str) -> None

Persist a terminal failure for this turn.

cancel

python
@transition(
    status,
    source=[
        TurnStatus.PENDING, TurnStatus.RUNNING, TurnStatus.AWAITING_APPROVAL
    ],
    target=TurnStatus.CANCELED,
    on_success=save_state,
)
def cancel() -> None

Cancel this turn without deleting its audit trail.

Released under the AGPL-3.0 License.