Skip to content

Models API Reference#

Core data models for the OpenIntent SDK.

Intent#

Intent dataclass #

Intent(
    id: str,
    title: str,
    description: str,
    version: int,
    status: IntentStatus,
    state: IntentState,
    constraints: dict[str, Any] = dict(),
    parent_intent_id: Optional[str] = None,
    depends_on: list[str] = list(),
    created_at: Optional[datetime] = None,
    updated_at: Optional[datetime] = None,
    created_by: Optional[str] = None,
    confidence: int = 0,
)

Core intent object representing a goal to be coordinated.

RFC-0002 Intent Graphs: Supports hierarchical parent-child relationships via parent_intent_id and dependency graphs via depends_on.

has_dependencies property #

has_dependencies: bool

Check if this intent depends on other intents.

has_parent property #

has_parent: bool

Check if this intent has a parent (is a child intent).

IntentEvent#

IntentEvent dataclass #

IntentEvent(
    id: str,
    intent_id: str,
    event_type: EventType,
    actor: Optional[str],
    payload: dict[str, Any],
    created_at: datetime,
    proof: Optional[EventProof] = None,
    event_hash: Optional[str] = None,
    previous_event_hash: Optional[str] = None,
    sequence: Optional[int] = None,
    trace_id: Optional[str] = None,
    parent_event_id: Optional[str] = None,
)

Immutable event in the intent's audit log.

RFC-0018: Optional proof field for signed events. RFC-0019: Optional event_hash, previous_event_hash, sequence for verifiable logs. RFC-0020: Optional trace_id, parent_event_id for distributed tracing.

IntentStatus#

IntentStatus #

Bases: str, Enum

Status of an intent in its lifecycle.

EventType#

EventType #

Bases: str, Enum

Types of events that can occur on an intent.

IntentLease#

IntentLease dataclass #

IntentLease(
    id: str,
    intent_id: str,
    agent_id: str,
    scope: str,
    status: LeaseStatus,
    expires_at: datetime,
    created_at: datetime,
)

Lease granting exclusive access to a scope within an intent.

LeaseStatus#

LeaseStatus #

Bases: str, Enum

Status of a scope lease.

IntentPortfolio#

IntentPortfolio dataclass #

IntentPortfolio(
    id: str,
    name: str,
    description: Optional[str],
    created_by: str,
    status: PortfolioStatus,
    metadata: dict[str, Any],
    governance_policy: dict[str, Any],
    created_at: Optional[datetime] = None,
    updated_at: Optional[datetime] = None,
    intents: list[Intent] = list(),
    aggregate_status: Optional[AggregateStatus] = None,
)

Collection of related intents with aggregate tracking and shared governance.

PortfolioStatus#

PortfolioStatus #

Bases: str, Enum

Status of a portfolio.

LLM Observability Models#

ToolCallPayload#

ToolCallPayload dataclass #

ToolCallPayload(
    tool_name: str,
    tool_id: str,
    arguments: dict[str, Any],
    provider: Optional[str] = None,
    model: Optional[str] = None,
    parent_request_id: Optional[str] = None,
    result: Optional[Any] = None,
    error: Optional[str] = None,
    duration_ms: Optional[int] = None,
    token_count: Optional[int] = None,
)

Structured payload for tool call events. Captures LLM-initiated tool/function calls with full context.

LLMRequestPayload#

LLMRequestPayload dataclass #

LLMRequestPayload(
    request_id: str,
    provider: str,
    model: str,
    messages_count: int,
    tools_available: list[str] = list(),
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    stream: bool = False,
    response_content: Optional[str] = None,
    tool_calls: list[dict[str, Any]] = list(),
    finish_reason: Optional[str] = None,
    prompt_tokens: Optional[int] = None,
    completion_tokens: Optional[int] = None,
    total_tokens: Optional[int] = None,
    duration_ms: Optional[int] = None,
    error: Optional[str] = None,
)

Structured payload for LLM request events. Captures the full context of an LLM API call.

StreamState#

StreamState dataclass #

StreamState(
    stream_id: str,
    intent_id: str,
    agent_id: str,
    status: StreamStatus,
    provider: str,
    model: str,
    chunks_received: int = 0,
    tokens_streamed: int = 0,
    started_at: Optional[datetime] = None,
    completed_at: Optional[datetime] = None,
    cancelled_at: Optional[datetime] = None,
    cancel_reason: Optional[str] = None,
    error: Optional[str] = None,
)

Tracks the state of a streaming operation. Used for real-time coordination and cancellation.

Task Decomposition (RFC-0012)#

Task#

Task dataclass #

Task(
    id: str,
    intent_id: str,
    name: str,
    version: int = 1,
    status: TaskStatus = TaskStatus.PENDING,
    plan_id: Optional[str] = None,
    description: Optional[str] = None,
    priority: str = "normal",
    input: dict[str, Any] = dict(),
    output: Optional[dict[str, Any]] = None,
    artifacts: list[str] = list(),
    assigned_agent: Optional[str] = None,
    lease_id: Optional[str] = None,
    capabilities_required: list[str] = list(),
    depends_on: list[str] = list(),
    blocks: list[str] = list(),
    parent_task_id: Optional[str] = None,
    retry_policy: Optional[str] = None,
    timeout_seconds: Optional[int] = None,
    attempt: int = 1,
    max_attempts: int = 3,
    permissions: str = "inherit",
    memory_policy: Optional[MemoryPolicy] = None,
    requires_tools: list[ToolRequirement] = list(),
    blocked_reason: Optional[str] = None,
    error: Optional[str] = None,
    metadata: dict[str, Any] = dict(),
    created_at: Optional[datetime] = None,
    started_at: Optional[datetime] = None,
    completed_at: Optional[datetime] = None,
)

A concrete, bounded unit of work derived from an intent (RFC-0012).

TaskStatus#

TaskStatus #

Bases: str, Enum

Status of a task in its lifecycle (RFC-0012).

Plan#

Plan dataclass #

Plan(
    id: str,
    intent_id: str,
    version: int = 1,
    state: PlanState = PlanState.DRAFT,
    tasks: list[str] = list(),
    checkpoints: list[Checkpoint] = list(),
    conditions: list[PlanCondition] = list(),
    on_failure: PlanFailureAction = PlanFailureAction.PAUSE_AND_ESCALATE,
    on_complete: str = "notify",
    metadata: dict[str, Any] = dict(),
    created_at: Optional[datetime] = None,
    updated_at: Optional[datetime] = None,
)

Execution strategy for achieving an intent (RFC-0012).

PlanState#

PlanState #

Bases: str, Enum

State of a plan (RFC-0012).

Coordinator Governance (RFC-0013)#

CoordinatorLease#

CoordinatorLease dataclass #

CoordinatorLease(
    id: str,
    intent_id: Optional[str] = None,
    portfolio_id: Optional[str] = None,
    agent_id: str = "",
    role: str = "coordinator",
    supervisor_id: Optional[str] = None,
    coordinator_type: CoordinatorType = CoordinatorType.LLM,
    scope: CoordinatorScope = CoordinatorScope.INTENT,
    status: CoordinatorStatus = CoordinatorStatus.ACTIVE,
    guardrails: Optional[Guardrails] = None,
    heartbeat_interval_seconds: int = 60,
    last_heartbeat: Optional[datetime] = None,
    granted_at: Optional[datetime] = None,
    expires_at: Optional[datetime] = None,
    version: int = 1,
    metadata: dict[str, Any] = dict(),
)

Lease granting coordinator authority over an intent or portfolio (RFC-0013).

CoordinatorType#

CoordinatorType #

Bases: str, Enum

Type of coordinator (RFC-0013).

CoordinatorStatus#

CoordinatorStatus #

Bases: str, Enum

Lifecycle state of a coordinator (RFC-0013).

DecisionRecord#

DecisionRecord dataclass #

DecisionRecord(
    id: str,
    coordinator_id: str,
    intent_id: str,
    decision_type: DecisionType,
    summary: str,
    rationale: str,
    alternatives_considered: list[dict[str, Any]] = list(),
    confidence: Optional[float] = None,
    timestamp: Optional[datetime] = None,
)

Auditable record of a coordination decision (RFC-0013).

DecisionType#

DecisionType #

Bases: str, Enum

Type of coordinator decision (RFC-0013).

Credential Vaults (RFC-0014)#

CredentialVault#

CredentialVault dataclass #

CredentialVault(
    id: str,
    owner_id: str,
    name: str,
    credentials: list[str] = list(),
    created_at: Optional[datetime] = None,
)

User-owned encrypted store for external service credentials (RFC-0014).

Credential#

Credential dataclass #

Credential(
    id: str,
    vault_id: str,
    service: str,
    label: str,
    auth_type: AuthType,
    scopes_available: list[str] = list(),
    status: CredentialStatus = CredentialStatus.ACTIVE,
    metadata: dict[str, Any] = dict(),
    created_at: Optional[datetime] = None,
    rotated_at: Optional[datetime] = None,
    expires_at: Optional[datetime] = None,
)

Encrypted record of authentication material for an external service (RFC-0014).

ToolGrant#

ToolGrant dataclass #

ToolGrant(
    id: str,
    credential_id: str,
    agent_id: str,
    granted_by: str,
    scopes: list[str] = list(),
    constraints: Optional[GrantConstraints] = None,
    source: GrantSource = GrantSource.DIRECT,
    delegatable: bool = False,
    delegation_depth: int = 0,
    delegated_from: Optional[str] = None,
    context: dict[str, Any] = dict(),
    status: GrantStatus = GrantStatus.ACTIVE,
    expires_at: Optional[datetime] = None,
    created_at: Optional[datetime] = None,
    revoked_at: Optional[datetime] = None,
)

Permission linking an agent to external service tools (RFC-0014).

ToolInvocation#

ToolInvocation dataclass #

ToolInvocation(
    invocation_id: str,
    grant_id: str,
    service: str,
    tool: str,
    agent_id: str,
    parameters: dict[str, Any] = dict(),
    status: InvocationStatus = InvocationStatus.SUCCESS,
    result: Optional[dict[str, Any]] = None,
    error: Optional[dict[str, Any]] = None,
    cost: Optional[dict[str, Any]] = None,
    duration_ms: Optional[int] = None,
    idempotency_key: Optional[str] = None,
    context: dict[str, Any] = dict(),
    timestamp: Optional[datetime] = None,
)

Record of a tool proxy invocation (RFC-0014).

AuthType#

AuthType #

Bases: str, Enum

Authentication type for a credential (RFC-0014).

GrantStatus#

GrantStatus #

Bases: str, Enum

Status of a tool grant (RFC-0014).

Agent Memory (RFC-0015)#

MemoryEntry#

MemoryEntry dataclass #

MemoryEntry(
    id: str,
    agent_id: str,
    namespace: str,
    key: str,
    value: dict[str, Any],
    memory_type: MemoryType,
    version: int = 1,
    scope: Optional[MemoryScope] = None,
    tags: list[str] = list(),
    ttl: Optional[str] = None,
    pinned: bool = False,
    priority: MemoryPriority = MemoryPriority.NORMAL,
    sensitivity: Optional[MemorySensitivity] = None,
    curated_by: Optional[str] = None,
    created_at: Optional[datetime] = None,
    updated_at: Optional[datetime] = None,
    expires_at: Optional[datetime] = None,
)

Fundamental unit of agent state (RFC-0015).

MemoryType#

MemoryType #

Bases: str, Enum

Memory tier for a memory entry (RFC-0015).

MemoryPriority#

MemoryPriority #

Bases: str, Enum

Priority level for episodic memory entries (RFC-0015).

EvictionPolicy#

EvictionPolicy #

Bases: str, Enum

Eviction policy for episodic memory (RFC-0015).

Agent Lifecycle (RFC-0016)#

AgentRecord#

AgentRecord dataclass #

AgentRecord(
    agent_id: str,
    status: AgentStatus = AgentStatus.ACTIVE,
    role_id: Optional[str] = None,
    name: Optional[str] = None,
    capabilities: list[str] = list(),
    capacity: Optional[AgentCapacity] = None,
    endpoint: Optional[str] = None,
    heartbeat_config: Optional[HeartbeatConfig] = None,
    metadata: dict[str, Any] = dict(),
    registered_at: Optional[datetime] = None,
    last_heartbeat_at: Optional[datetime] = None,
    drain_timeout_seconds: Optional[int] = None,
    version: int = 1,
    public_key: Optional[str] = None,
    did: Optional[str] = None,
    key_algorithm: Optional[str] = None,
    key_registered_at: Optional[datetime] = None,
    key_expires_at: Optional[datetime] = None,
    previous_keys: list[str] = list(),
)

Protocol-level representation of a participating agent (RFC-0016, RFC-0018).

AgentStatus#

AgentStatus #

Bases: str, Enum

Lifecycle status of an agent (RFC-0016).

Triggers (RFC-0017)#

Trigger#

Trigger dataclass #

Trigger(
    trigger_id: str,
    name: str,
    type: TriggerType,
    enabled: bool = True,
    condition: Optional[TriggerCondition] = None,
    intent_template: Optional[IntentTemplate] = None,
    deduplication: DeduplicationMode = DeduplicationMode.ALLOW,
    namespace: Optional[str] = None,
    fire_count: int = 0,
    version: int = 1,
    created_at: Optional[datetime] = None,
    updated_at: Optional[datetime] = None,
    last_fired_at: Optional[datetime] = None,
)

Standing declaration that creates intents when a condition is met (RFC-0017).

TriggerType#

TriggerType #

Bases: str, Enum

Type of trigger (RFC-0017).

TriggerState#

TriggerState #

Bases: str, Enum

Lifecycle state of a trigger (RFC-0017).

DeduplicationMode#

DeduplicationMode #

Bases: str, Enum

Deduplication behavior when a trigger fires (RFC-0017).

Access Control (RFC-0011)#

AccessPolicy#

AccessPolicy #

Bases: str, Enum

Default access policy when no ACL entry matches (RFC-0011).

Permission#

Permission #

Bases: str, Enum

Permission level for intent access control (RFC-0011).

AccessRequestStatus#

AccessRequestStatus #

Bases: str, Enum

Status of an access request (RFC-0011).

Cost & Retry#

CostType#

CostType #

Bases: str, Enum

Type of cost/resource being tracked.

RetryStrategy#

RetryStrategy #

Bases: str, Enum

Retry strategy for failure handling.