Public API for the openclaw package.
class BaseAgent(ABC):
name: str # Unique agent identifier (required class attribute)
description: str # Human-readable description
capabilities: list[str] # Routing capability tags
@abstractmethod
def can_handle(self, task: str) -> float:
"""Return confidence score 0.0–1.0 for handling this task."""
@abstractmethod
def get_tools(self) -> list[ToolDefinition]:
"""Return the tools this agent exposes to Claude."""
@abstractmethod
async def run(self, task: str, session: Session) -> str:
"""Execute the task and return the result string."""
def on_register(self) -> None:
"""Hook called when registered in AgentRegistry."""
def on_unregister(self) -> None:
"""Hook called when removed from AgentRegistry."""
def info(self) -> dict[str, Any]:
"""Return serialisable summary of this agent."""class AgentRegistry:
def register(self, agent: BaseAgent, plugin_name: str | None = None) -> None
def unregister(self, name: str) -> BaseAgent
def unregister_by_plugin(self, plugin_name: str) -> list[str]
def get(self, name: str) -> BaseAgent # raises AgentNotFoundError
def all(self) -> list[BaseAgent]
def by_capability(self, capability: str) -> list[BaseAgent]
def route(self, task: str, top_n: int = 3) -> list[tuple[BaseAgent, float]]
def summary(self) -> list[dict[str, Any]]
def __contains__(self, name: str) -> bool
def __len__(self) -> int@tool(
name: str | None = None, # defaults to function name
description: str | None = None, # defaults to docstring
parameters: dict | None = None, # defaults to inferred from type hints
)
def my_tool(param: str) -> str: ...@dataclass
class ToolDefinition:
name: str
description: str
parameters: dict[str, Any] # JSON Schema
fn: Callable
source_plugin: str | None # set when registered by a plugin
def to_api_dict(self) -> dict[str, Any] # Anthropic API format
async def execute(self, **kwargs) -> Anyclass ToolRegistry:
def register(self, tool_def: ToolDefinition) -> None
def register_fn(self, fn: Callable, source_plugin: str | None = None) -> ToolDefinition
def register_module(self, module: Any, source_plugin: str | None = None) -> list[str]
def get(self, name: str) -> ToolDefinition # raises ToolNotFoundError
def all(self) -> list[ToolDefinition]
def by_names(self, names: list[str]) -> list[ToolDefinition]
def unregister_by_plugin(self, plugin_name: str) -> list[str]
def to_api_list(self, names: list[str] | None = None) -> list[dict]
async def execute(self, name: str, **kwargs) -> Any
def __contains__(self, name: str) -> bool
def __len__(self) -> int@dataclass
class Session:
session_id: str # UUID
conversation: Conversation
shared_state: SharedState
created_at: float
metadata: dict[str, Any]
def reset_conversation(self) -> None
def age_seconds(self) -> float
def new_session() -> Session # factory via app.new_session()class SharedState:
def set(self, key: str, value: Any, ttl: int | None = None) -> None
def get(self, key: str, default: Any = None) -> Any
def delete(self, key: str) -> None
def keys(self) -> list[str]
def clear(self) -> None
def __contains__(self, key: str) -> boolclass Role(str, Enum):
USER = "user"
ASSISTANT = "assistant"
SYSTEM = "system"
TOOL = "tool"@dataclass
class Message:
role: Role
content: str | list[dict]
name: str | None
metadata: dict
@classmethod def user(cls, text: str) -> Message
@classmethod def assistant(cls, text: str) -> Message
@classmethod def system(cls, text: str) -> Message
def to_api_dict(self) -> dictclass Conversation:
max_history: int
def add(self, message: Message) -> None
def add_user(self, text: str) -> None
def add_assistant(self, text: str) -> None
def to_api_list(self) -> list[dict]
def clear(self) -> None
def __len__(self) -> int@dataclass
class PluginManifest:
name: str
version: str
description: str
author: str
plugin_type: PluginType # AGENT | TOOL | PROVIDER | EXTENSION
entry_point: str
dependencies: list[str]
capabilities: list[str]
min_openclaw_version: str
homepage: str
license: str
@classmethod def from_dict(cls, data: dict) -> PluginManifest
def to_dict(self) -> dictclass BasePlugin(ABC):
manifest: PluginManifest
status: PluginStatus
@abstractmethod
def on_load(self, agent_registry: AgentRegistry, tool_registry: ToolRegistry) -> None
@abstractmethod
def on_unload(self) -> None
def get_agents(self) -> list[BaseAgent] # override to add agents
def get_tools(self) -> list[ToolDefinition] # override to add toolsclass PluginManager:
def __init__(
self,
plugins_dir: str | Path,
agent_registry: AgentRegistry,
tool_registry: ToolRegistry,
)
def discover(self) -> list[Path]
def load_all(self) -> dict[str, bool] # name → success
def load_plugin(self, plugin_path: str | Path) -> BasePlugin
def unload_plugin(self, plugin_name: str) -> bool
def reload_plugin(self, plugin_name: str) -> BasePlugin | None
def get(self, plugin_name: str) -> BasePlugin | None
def all(self) -> list[BasePlugin]
def is_loaded(self, plugin_name: str) -> bool
def summary(self) -> list[dict[str, Any]]
def __len__(self) -> intclass PluginWatcher:
def __init__(
self,
plugins_dir: str | Path,
on_plugin_added: Callable[[Path], None],
on_plugin_removed: Callable[[str], None],
poll_interval: float = 1.0,
)
def start(self) -> None # starts watchdog or polling fallback
def stop(self) -> None
def rescan(self) -> None # manual scan (useful in tests)class AnthropicProvider(BaseProvider):
def __init__(
self,
api_key: str | None = None, # defaults to ANTHROPIC_API_KEY env
model: str | None = None, # defaults to OPENCLAW_MODEL env or claude-sonnet-4-6
max_tokens: int | None = None,
)
@property
def model_id(self) -> str
async def run_agent_loop(
self,
*,
system_prompt: str,
messages: list[dict],
tools: list[dict],
tool_executor: Any, # must have: async execute(name, **kwargs)
max_iterations: int = 15,
) -> str
async def complete(
self,
*,
system_prompt: str,
messages: list[dict],
max_tokens: int = 2048,
) -> strdef create_app(
plugins_dir: str | Path | None = None,
config_dir: str | Path | None = None,
enable_plugin_watcher: bool = True,
) -> App
@dataclass
class App:
agent_registry: AgentRegistry
tool_registry: ToolRegistry
plugin_manager: PluginManager
plugin_watcher: PluginWatcher | None
orchestrator: OrchestratorAgent
settings: dict
def new_session(self) -> Session
async def run(self, task: str, session: Session | None = None) -> str
def stop(self) -> None| Exception | Description |
|---|---|
OpenClawError |
Base class for all errors |
AgentNotFoundError(name) |
Agent not in registry |
AgentExecutionError(agent_name, reason) |
Agent run() failed |
ToolNotFoundError(name) |
Tool not in registry |
ToolExecutionError(tool_name, reason) |
Tool execution failed |
PluginLoadError(path, reason) |
Plugin failed to import or initialise |
PluginManifestError(path, reason) |
plugin.yaml missing or invalid |
PluginConflictError(name, existing, new) |
Duplicate agent/tool name |
ProviderAuthError |
API key missing or invalid |
ProviderRateLimitError |
Rate limit hit |
MaxIterationsError(max) |
Agentic loop exceeded limit |
GitError |
Local git operation failed |
GitHubAPIError(status_code, message) |
GitHub API error |
GitLabAPIError(status_code, message) |
GitLab API error |
WSL2Error |
WSL2 operation failed |
WSL2DistroNotFoundError(distro) |
WSL2 distro not installed |