Files

T

HAL9000 acaea2ba1c docs: add API documentation (CLI reference, Python API, protocols) [AUTO-DOCS-2]

2026-04-28 09:25:53 +00:00

16 KiB

Raw Blame History

Python API — Application Layer

This page documents the application layer of the CleverAgents Python package: application services, the dependency injection (DI) container, domain models, and key protocols. For module-level API documentation (exceptions, registries, adapters) see the Module Index.

See ADR-001 (layered architecture) and ADR-003 (dependency injection) for design rationale.

Architecture Overview

CleverAgents follows a strict layered architecture:

Entry Points (CLI / TUI / A2A server)
    ↓
Application Layer  ← this page
    ↓
Domain Layer       (models, value objects, domain services)
    ↓
Infrastructure     (database, file system, external services)
    ↓
Integration        (LangChain/LangGraph, MCP, LSP adapters)
    ↓
Core               (exceptions, retry, circuit breaker, async cleanup)

All cross-layer dependencies flow downward only. The DI container wires everything together at startup.

Dependency Injection Container

Module: cleveragents.application.container

CleverAgents uses dependency-injector for IoC. The CleverAgentsContainer is the root container; all application services are registered as providers and resolved lazily.

from cleveragents.application.container import CleverAgentsContainer

container = CleverAgentsContainer()
container.config.from_dict({"provider": "openai", ...})
container.wire(modules=[__name__])

# Resolve services
plan_service = container.plan_service()
registry_service = container.registry_service()
session_service = container.session_service()
invariant_service = container.invariant_service()

Key Providers

Provider	Type	Description
`container.settings`	`Singleton`	Application `Settings` instance
`container.event_bus`	`Singleton`	In-process event bus
`container.plan_service`	`Singleton`	`PlanLifecycleService`
`container.registry_service`	`Singleton`	`RegistryService`
`container.session_service`	`Singleton`	`SessionService`
`container.invariant_service`	`Singleton`	`InvariantService`
`container.actor_registry`	`Singleton`	`ActorRegistry`
`container.tool_registry`	`Singleton`	`ToolRegistry`
`container.skill_registry`	`Singleton`	`SkillRegistry`
`container.provider_registry`	`Singleton`	`ProviderRegistry`
`container.mcp_registry`	`Singleton`	`McpRegistry`
`container.lsp_registry`	`Singleton`	`LSPRegistry`
`container.resource_registry`	`Singleton`	`ResourceRegistry`
`container.a2a_facade`	`Singleton`	`A2aLocalFacade`

Tip: Always obtain services via the container rather than constructing them directly. This ensures correct lifecycle management and dependency resolution.

PlanLifecycleService

Module: cleveragents.application.services.plan_service

The central application service for the plan lifecycle. Orchestrates the four phases (Action → Strategize → Execute → Apply) and coordinates actors, tools, resources, and invariants.

from cleveragents.application.services.plan_service import PlanLifecycleService

service: PlanLifecycleService = container.plan_service()

# Create a plan from an action
plan = await service.create_plan(
    action_name="local/refactor",
    project_names=["my-project"],
    args={"target": "src/"},
    automation_profile="review",
)

# Execute the plan (Strategize + Execute phases)
result = await service.execute_plan(plan.plan_id)

# Apply the sandbox changeset
await service.apply_plan(plan.plan_id)

Key Methods

Method	Returns	Description
`create_plan(action_name, project_names, args, ...)`	`Plan`	Instantiate a plan from an action
`execute_plan(plan_id)`	`PlanExecutionResult`	Run Strategize and Execute phases
`apply_plan(plan_id)`	`PlanApplyResult`	Merge sandbox changeset into real resources
`get_plan(plan_id)`	`Plan`	Retrieve a plan by ID
`list_plans(filters)`	`list[Plan]`	List plans with optional filters
`cancel_plan(plan_id, reason)`	`Plan`	Cancel a running plan
`rollback_plan(plan_id, checkpoint_id)`	`Plan`	Roll back to a checkpoint
`get_plan_diff(plan_id)`	`PlanDiff`	Get the sandbox diff for a plan
`correct_decision(decision_id, mode, guidance)`	`CorrectionAttempt`	Correct a decision and recompute
`get_decision_tree(plan_id)`	`DecisionTree`	Retrieve the full decision tree

RegistryService

Module: cleveragents.application.services.registry_service

Unified CRUD service for all registry entities: actions, actors, skills, tools, resources, projects, LSP servers, and automation profiles.

from cleveragents.application.services.registry_service import RegistryService

service: RegistryService = container.registry_service()

# Register an action from a YAML file
action = await service.register_action(Path("examples/actions/refactor.yaml"))

# List all tools
tools = await service.list_tools(namespace="local", source="mcp")

# Register a resource
resource = await service.register_resource(
    type_name="git-checkout",
    name="local/my-repo",
    args={"url": "https://github.com/org/repo.git"},
)

Key Methods

Method	Description
`register_action(path, update=False)`	Register or update an action from YAML
`list_actions(namespace, state, regex)`	List actions
`get_action(name)`	Get an action by name
`archive_action(name)`	Archive an action
`register_actor(path, update=False)`	Register or update an actor from YAML
`list_actors()`	List all actors
`get_actor(name)`	Get an actor by name
`register_skill(path, update=False)`	Register or update a skill from YAML
`list_skills(namespace, source)`	List skills
`register_tool(path, update=False)`	Register or update a tool from YAML
`list_tools(namespace, source, type)`	List tools
`register_resource(type_name, name, args)`	Register a resource
`list_resources(type, include_stopped)`	List resources
`get_resource(name)`	Get a resource by name
`create_project(name, description, resources, invariants)`	Create a project
`list_projects(namespace, regex)`	List projects
`link_resource_to_project(project, resource, read_only)`	Link a resource to a project

SessionService

Module: cleveragents.application.services.session_service

Manages conversation sessions — creation, retrieval, message history, export, and import.

from cleveragents.application.services.session_service import SessionService

service: SessionService = container.session_service()

# Create a session
session = await service.create_session(actor="openai/gpt-4o")

# Send a message
response = await service.send_message(
    session_id=session.session_id,
    content="What is the current plan status?",
)

# Export as JSON
export_data = await service.export_session(session.session_id)

# Export as Markdown transcript
md = await service.export_session_markdown(session.session_id)

Key Methods

Method	Description
`create_session(actor)`	Create a new session
`get_session(session_id)`	Retrieve a session
`list_sessions()`	List all sessions
`delete_session(session_id)`	Delete a session
`send_message(session_id, content, stream)`	Send a message and get a response
`export_session(session_id)`	Export session as a JSON-serializable dict
`export_session_markdown(session_id)`	Export session as a Markdown transcript
`import_session(data)`	Import a session from a previously exported dict

InvariantService

Module: cleveragents.application.services.invariant_service

See cleveragents.core — Invariant Service for the full API reference. The service is registered as a Singleton in the DI container:

invariant_service = container.invariant_service()

Domain Models

Module: cleveragents.domain.models

All domain models inherit from DomainBaseModel (see Core Utilities).

Plan

from cleveragents.domain.models.core.plan import Plan, PlanPhase, PlanState

class Plan(DomainBaseModel):
    plan_id: str              # ULID
    name: str | None          # namespaced name (top-level plans only)
    action_name: str          # action this plan was instantiated from
    project_names: list[str]  # bound projects
    phase: PlanPhase          # Action | Strategize | Execute | Apply
    state: PlanState          # running | applied | errored | cancelled | ...
    automation_profile: str
    created_at: datetime
    updated_at: datetime
    parent_plan_id: str | None

PlanPhase enum: ACTION, STRATEGIZE, EXECUTE, APPLY

PlanState enum: RUNNING, APPLIED, CONSTRAINED, ERRORED, CANCELLED

Action

from cleveragents.domain.models.core.action import Action

class Action(DomainBaseModel):
    name: str                    # namespaced name
    description: str
    definition_of_done: str
    strategy_actor: str | None
    execution_actor: str | None
    estimation_actor: str | None
    invariant_actor: str | None
    args: list[ActionArgument]
    invariants: list[str]
    state: str                   # "active" | "archived"

Decision

from cleveragents.domain.models.core.decision import Decision, DecisionType

class Decision(DomainBaseModel):
    decision_id: str        # ULID
    plan_id: str
    type: DecisionType
    question: str
    chosen_option: str
    alternatives: list[str]
    confidence: float       # 0.0–1.0
    rationale: str
    phase: PlanPhase
    created_at: datetime
    superseded_by: str | None

DecisionType enum: PROMPT_DEFINITION, INVARIANT_ENFORCED, STRATEGY_CHOICE, SUBPLAN_SPAWN, SUBPLAN_PARALLEL_SPAWN, …

Resource

from cleveragents.domain.models.core.resource import Resource

class Resource(DomainBaseModel):
    resource_id: str        # ULID
    name: str               # namespaced name
    type_name: str          # e.g. "git-checkout", "sqlite"
    description: str | None
    is_physical: bool
    parent_ids: list[str]
    child_ids: list[str]
    state: str              # "active" | "stopped"
    created_at: datetime

Project

from cleveragents.domain.models.core.project import Project

class Project(DomainBaseModel):
    name: str               # namespaced name (sole identifier — no ULID)
    description: str | None
    resource_links: list[ResourceLink]
    invariants: list[str]
    context_policy: ContextPolicy | None
    created_at: datetime

Session

from cleveragents.domain.models.core.session import Session

class Session(DomainBaseModel):
    session_id: str
    actor: str
    messages: list[Message]
    linked_plan_ids: list[str]
    created_at: datetime
    updated_at: datetime

    def as_export_markdown(self) -> str:
        """Render a human-readable Markdown transcript (lossy, not importable)."""

Invariant

from cleveragents.domain.models.core.invariant import Invariant, InvariantScope

class Invariant(DomainBaseModel):
    invariant_id: str       # ULID
    text: str
    scope: InvariantScope   # GLOBAL | PROJECT | ACTION | PLAN
    source_name: str        # project/action/plan name, or "global"
    active: bool
    non_overridable: bool   # global invariants only
    created_at: datetime

InvariantScope enum: GLOBAL, PROJECT, ACTION, PLAN

Key Protocols and Interfaces

`AIProviderInterface`

Module: cleveragents.providers.interface

Protocol that all AI provider implementations must satisfy.

class AIProviderInterface(Protocol):
    @property
    def name(self) -> str: ...

    @property
    def model_id(self) -> str: ...

    async def generate(
        self,
        messages: list[Message],
        tools: list[ToolSpec] | None = None,
        stream: bool = False,
    ) -> GenerationResult: ...

`ResourceHandler`

Module: cleveragents.resource.handlers.base

Protocol for resource handler implementations.

class ResourceHandler(Protocol):
    async def read(self, resource_id: str, context: ...) -> Any: ...
    async def write(self, resource_id: str, data: Any, context: ...) -> None: ...
    async def delete(self, resource_id: str, context: ...) -> None: ...
    async def list_children(self, resource_id: str, context: ...) -> list[str]: ...
    async def diff(self, resource_id: str, other_id: str, context: ...) -> str: ...
    async def checkpoint(self, resource_id: str) -> str: ...
    async def rollback(self, resource_id: str, checkpoint_id: str) -> None: ...
    async def create_sandbox(self, resource_id: str) -> "Sandbox": ...

`LLMCaller`

Module: cleveragents.tool

Protocol for calling an LLM. Implement this to plug in a custom provider into the tool-calling runtime.

class LLMCaller(Protocol):
    async def call(
        self,
        messages: list[Message],
        tools: list[ToolSpec],
    ) -> LLMResponse: ...

End-to-End Example

The following example shows how to use the Python API to create and execute a plan programmatically:

import asyncio
from cleveragents.application.container import CleverAgentsContainer

async def main():
    # Bootstrap the container
    container = CleverAgentsContainer()
    container.config.from_env("CLEVERAGENTS_")

    plan_service = container.plan_service()
    registry_service = container.registry_service()

    # Register a resource
    await registry_service.register_resource(
        type_name="git-checkout",
        name="local/my-repo",
        args={"url": "https://github.com/org/repo.git", "path": "/workspace/repo"},
    )

    # Create a project
    await registry_service.create_project(
        name="my-project",
        description="Main application project",
        resources=["local/my-repo"],
    )

    # Instantiate a plan
    plan = await plan_service.create_plan(
        action_name="local/refactor",
        project_names=["my-project"],
        args={"target": "src/"},
        automation_profile="review",
    )
    print(f"Plan created: {plan.plan_id}")

    # Execute the plan
    result = await plan_service.execute_plan(plan.plan_id)
    print(f"Execution result: {result.state}")

    # Review the decision tree
    tree = await plan_service.get_decision_tree(plan.plan_id)
    for decision in tree.decisions:
        print(f"  [{decision.type}] {decision.question[:60]}…")

    # Apply if satisfied
    await plan_service.apply_plan(plan.plan_id)
    print("Plan applied successfully.")

asyncio.run(main())

Event Bus

Module: cleveragents.application.events

The in-process event bus enables loose coupling between application services. Services emit events; subscribers react without direct dependencies.

from cleveragents.application.events import EventBus, EventType

bus: EventBus = container.event_bus()

# Subscribe to plan phase transitions
@bus.on(EventType.PLAN_PHASE_CHANGED)
async def on_phase_change(event):
    print(f"Plan {event.plan_id} moved to {event.new_phase}")

# Emit an event (done internally by services)
await bus.emit(EventType.PLAN_PHASE_CHANGED, plan_id="01JXYZ...", new_phase="execute")

Key Event Types

Event	Description
`PLAN_CREATED`	A new plan was instantiated
`PLAN_PHASE_CHANGED`	A plan advanced to a new phase
`PLAN_APPLIED`	A plan was successfully applied
`PLAN_CANCELLED`	A plan was cancelled
`PLAN_ERRORED`	A plan encountered an unrecoverable error
`DECISION_RECORDED`	A new decision was added to the decision tree
`INVARIANT_VIOLATED`	An invariant was not satisfied
`INVARIANT_ENFORCED`	An invariant enforcement record was created
`INVARIANT_RECONCILED`	Invariant reconciliation completed for a phase
`TOOL_EXECUTED`	A tool invocation completed
`SESSION_CREATED`	A new session was created
`SESSION_MESSAGE_ADDED`	A message was added to a session

16 KiB Raw Blame History Unescape Escape

Python API — Application Layer

Architecture Overview

Dependency Injection Container

Key Providers

PlanLifecycleService

Key Methods

RegistryService

Key Methods

SessionService

Key Methods

InvariantService

Domain Models

Plan

Action

Decision

Resource

Project

Session

Invariant

Key Protocols and Interfaces

AIProviderInterface

ResourceHandler

LLMCaller

End-to-End Example

Event Bus

Key Event Types

16 KiB

Raw Blame History

`AIProviderInterface`

`ResourceHandler`

`LLMCaller`