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.