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
class InferenceModelUse(models.TextChoices)What an inference model is used for (mirrors the LLM catalogue's model use).
InferenceModelStatus
class InferenceModelStatus(models.TextChoices)Lifecycle of a model in a provider's catalogue.
MCPPlacement
class MCPPlacement(models.TextChoices)Where an MCP server runs relative to the platform.
MCPTransport
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
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
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
class SessionStatus(models.TextChoices)Display projection of one persisted agent session's workflow state.
TurnStatus
class TurnStatus(models.TextChoices)Execution state of one prompt-to-response turn.
InferenceProvider
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
class Meta()Django model options for the inference provider child model.
__str__
def __str__() -> strReturn the provider's display label.
backend
@property
def backend() -> InferenceBackendReturn 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
def refresh_models() -> intRe-list this provider's models into :class:InferenceModel rows.
chat
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) -> InferenceResponseSend one non-streaming chat request through this provider.
InferenceModelManager
class InferenceModelManager(AngeeManager)Manager owning the upsert of model rows from a provider's catalogue.
sync_from_provider
def sync_from_provider(provider: Any) -> intUpsert 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
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
class Meta()Django model options for catalogue models.
__str__
def __str__() -> strReturn the model's display label.
provider_model_name
@property
def provider_model_name() -> strReturn the provider-native model id for runtimes that talk to the provider directly.
credential
@property
def credential() -> AnyReturn the API credential for this model, via its provider's integration.
chat
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) -> InferenceResponseSend one non-streaming chat request through this catalogue model.
SkillManager
class SkillManager(AngeeManager)Manager owning the reconcile of skill rows from a skill source.
sync_from_source
def sync_from_source(source: Any) -> intWalk the source for SKILL.md and upsert/prune :class:Skill rows.
Skill
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
class Meta()Django model options for discovered skills.
__str__
def __str__() -> strReturn the skill's display label.
MCPServer
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
class Meta()Django model options for MCP servers.
__str__
def __str__() -> strReturn the server's name.
builtin
@property
def builtin() -> strReturn the built-in MCP server key this row targets, if any.
resolved_url
@property
def resolved_url() -> strReturn 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
@property
def is_addressable() -> boolWhether 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
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
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
class Meta()Django model options for MCP tools.
__str__
def __str__() -> strReturn the tool's name.
AgentManager
class AgentManager(AngeeManager)Manager owning service-user lifecycle for agent principals.
service_username
def service_username(agent: Any) -> strReturn the deterministic username for agent's service user.
sync_service_user
def sync_service_user(agent: Any) -> AnyCreate 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
def deactivate_service_user(agent: Any, *, using: str | None = None) -> NoneDeactivate agent's linked service user, leaving attribution FKs intact.
Agent
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
class Meta()Django model options for agents.
__str__
def __str__() -> strReturn the agent's name.
save
def save(*args: Any, **kwargs: Any) -> NonePersist 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
def principal_subject() -> SubjectRefReturn 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
@property
def runtime_backend() -> AgentRuntimeReturn 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
@property
def can_provision() -> boolWhether the provision action may start from the current lifecycle facts.
can_deprovision
@property
def can_deprovision() -> boolWhether the teardown action is meaningful for the current rendered state.
can_delete
@property
def can_delete() -> boolWhether deleting the definition can leave no orphaned operator instance.
delete_blocker
def delete_blocker() -> str | NoneReturn the delete-blocking reason, or None when deletion is allowed.
mark_provisioning
@transition(
lifecycle,
source=[
AgentLifecycle.DRAFT,
AgentLifecycle.PROVISIONING,
AgentLifecycle.READY,
AgentLifecycle.DEPROVISIONED,
],
target=AgentLifecycle.PROVISIONING,
on_success=save_state,
)
def mark_provisioning() -> NoneEnter the provision flow: lifecycle provisioning, run state reset to stopped.
mark_workspace_provisioned
@transition(
lifecycle,
source=AgentLifecycle.PROVISIONING,
target=AgentLifecycle.PROVISIONING,
on_success=save_state,
)
def mark_workspace_provisioned(*, workspace: str) -> NoneRecord the workspace as soon as the operator creates it.
mark_service_provisioned
@transition(
lifecycle,
source=AgentLifecycle.PROVISIONING,
target=AgentLifecycle.PROVISIONING,
on_success=save_state,
)
def mark_service_provisioned(*, service: str) -> NoneRecord the service as soon as the operator creates it.
mark_provisioned
@transition(
lifecycle,
source=AgentLifecycle.PROVISIONING,
target=AgentLifecycle.READY,
on_success=save_state,
)
def mark_provisioned(*, workspace: str, service: str = "") -> NoneRecord 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
@transition(
lifecycle,
source=[
AgentLifecycle.DRAFT, AgentLifecycle.DEPROVISIONING,
AgentLifecycle.DEPROVISIONED
],
target=AgentLifecycle.DEPROVISIONED,
on_success=save_state,
)
def mark_deprovisioned() -> NoneClear the operator instance after teardown: lifecycle deprovisioned, run state stopped.
mark_deprovisioning
@transition(
lifecycle,
source=[
AgentLifecycle.DRAFT,
AgentLifecycle.PROVISIONING,
AgentLifecycle.READY,
AgentLifecycle.DEPROVISIONING,
AgentLifecycle.DEPROVISIONED,
],
target=AgentLifecycle.DEPROVISIONING,
on_success=save_state,
)
def mark_deprovisioning() -> NoneMark the agent as tearing down through the operator teardown flow.
mark_provision_failed
def mark_provision_failed(message: str,
*,
clear_instances: bool = False,
clear_service: bool = False) -> NoneRecord 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
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
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
def service_model_handle() -> strReturn the selected model handle in this agent's runtime convention.
mcp_config
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
def mcp_secret_name(server: MCPServer) -> strReturn 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
def mcp_bearer_env(server: MCPServer) -> strReturn 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
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
def inference_secret_name() -> strReturn 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
def inference_secret() -> strReturn 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
def provision_inference_secret() -> strReturn 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
def inference_credential_ready() -> boolWhether 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
def inference_credential_for_runtime() -> AnyReturn 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
class AgentSession(SqidMixin, AuditMixin, AngeeModel)Runtime-neutral persisted conversation backed by one workflow run.
Meta
class Meta()Django model options for persisted agent sessions.
__str__
def __str__() -> strReturn the session title or its agent name.
mark_running
@transition(
status,
source=[
SessionStatus.IDLE, SessionStatus.RUNNING,
SessionStatus.AWAITING_APPROVAL, SessionStatus.ERROR
],
target=SessionStatus.RUNNING,
on_success=save_state,
)
def mark_running() -> NoneProject active turn execution onto the session.
mark_idle
@transition(
status,
source=[SessionStatus.IDLE, SessionStatus.RUNNING, SessionStatus.ERROR],
target=SessionStatus.IDLE,
on_success=save_state,
)
def mark_idle() -> NoneProject a parked session awaiting its next user turn.
mark_awaiting_approval
@transition(
status,
source=SessionStatus.RUNNING,
target=SessionStatus.AWAITING_APPROVAL,
on_success=save_state,
)
def mark_awaiting_approval() -> NoneProject a suspended tool approval onto the session.
close
@transition(
status,
source=[
SessionStatus.IDLE, SessionStatus.RUNNING,
SessionStatus.AWAITING_APPROVAL, SessionStatus.ERROR
],
target=SessionStatus.CLOSED,
on_success=save_state,
)
def close() -> NoneClose the conversation so its workflow step can finish.
mark_error
@transition(
status,
source=[
SessionStatus.IDLE, SessionStatus.RUNNING,
SessionStatus.AWAITING_APPROVAL, SessionStatus.ERROR
],
target=SessionStatus.ERROR,
on_success=save_state,
)
def mark_error(message: str) -> NoneProject a session-level runtime error.
AgentTurn
class AgentTurn(SqidMixin, AuditMixin, AngeeModel)One prompt-to-response cycle with append-only ACP updates.
Meta
class Meta()Django model options for agent turns.
mark_running
@transition(
status,
source=[
TurnStatus.PENDING, TurnStatus.RUNNING, TurnStatus.AWAITING_APPROVAL
],
target=TurnStatus.RUNNING,
on_success=save_state,
)
def mark_running() -> NoneClaim or resume this turn for runtime execution.
mark_awaiting_approval
@transition(
status,
source=TurnStatus.RUNNING,
target=TurnStatus.AWAITING_APPROVAL,
on_success=save_state,
)
def mark_awaiting_approval() -> NoneSuspend this turn for deferred tool approval.
mark_completed
@transition(
status,
source=[TurnStatus.RUNNING, TurnStatus.AWAITING_APPROVAL],
target=TurnStatus.COMPLETED,
on_success=save_state,
)
def mark_completed(*, text: str, usage: Mapping[str, int]) -> NonePersist the final response projection for a completed turn.
mark_failed
@transition(
status,
source=[TurnStatus.RUNNING, TurnStatus.AWAITING_APPROVAL],
target=TurnStatus.FAILED,
on_success=save_state,
)
def mark_failed(message: str) -> NonePersist a terminal failure for this turn.
cancel
@transition(
status,
source=[
TurnStatus.PENDING, TurnStatus.RUNNING, TurnStatus.AWAITING_APPROVAL
],
target=TurnStatus.CANCELED,
on_success=save_state,
)
def cancel() -> NoneCancel this turn without deleting its audit trail.