Agent Openclaw¶

Agent OpenClaw — mvp.agent_openclaw

Cluster: Agents & LLM | Type: component | MCP Tools: 29

Overview¶

Channel gateway block with SQLite WAL-mode persistence for typed communication channels. Provides 8 operations — connect/disconnect/list/status channels and send/receive/broadcast/configure messages — backed by durable storage. Enforces TOS constraints identical to agent_nanoclaw: Anthropic and Claude Code providers are blocked; only ollama and openrouter are accepted.

When to use:

Managing typed communication channels (direct, pub-sub, webhook) between agents with persistent history
Broadcasting control signals across all active channels in a multi-agent deployment
Routing messages between heterogeneous G6 components via a unified channel abstraction

Example:

from mvp.agent_openclaw import AgentOpenclawBlock, ChannelInput

block = AgentOpenclawBlock(name="openclaw")
block.infer(ChannelInput(op="connect", channel_id="ch-01", channel_type="direct"))
result = block.infer(ChannelInput(
    op="send", channel_id="ch-01", message="Solver finished", provider="ollama",
))
# result.ok → True; result.value.success → True

Works well with: agent_nanoclaw, hat_orchestrator, autonomous_orchestrator

Public API¶

`AgentOpenclawBlock(LifecycleMixin, AIBlock[ChannelInput, ChannelOutput, None])`¶

Channel gateway with SQLite WAL persistence.

Field	Type	Default
`name`	`str`	'agent_openclaw'
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`stop() -> Result[None]`¶

`infer(data: ChannelInput) -> Result[ChannelOutput]`¶

`ChannelInput(BaseModel)`¶

Input to AgentOpenclawBlock — 8 basic channel operations.

Field	Type	Default
`op`	`Literal['connect', 'disconnect', 'list_channels', 'send', 'receive', 'broadcast', 'status', 'configure']`	required
`channel_type`	`str`	''
`channel_id`	`str`	''
`message`	`str`	''
`target`	`str`	''
`config`	`dict[str, Any]`	Field(default_factory=dict)
`session_id`	`str`	''
`provider`	`str`	'ollama'

`ChannelOutput(BaseModel)`¶

Output from AgentOpenclawBlock.

Field	Type	Default
`op`	`str`	required
`channel_id`	`str`	''
`success`	`bool`	False
`message`	`str`	''
`data`	`dict[str, Any]`	Field(default_factory=dict)
`channels`	`list[dict[str, Any]]`	Field(default_factory=list)
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`str`	'qualified-draft'
`mode`	`str`	'control_plane'
`backend`	`str`	''
`runtime_available`	`bool`	False
`runtime_result`	`dict[str, Any]`	Field(default_factory=dict)
`audit_id`	`str`	''

`AgentOpenclawMCPBlock(AIBlock[MCPOpenclawInput, MCPOpenclawOutput, dict])`¶

27-op channel gateway MCP block with SQLite persistence (deterministic).

Field	Type	Default
`name`	`str`	'agent_openclaw_mcp'
`state`	`dict`	field(default_factory=dict)
`db_path`	`str`	field(default_factory=lambda: os.environ.get('OPENCLAW_DB_PATH', _DEFAULT_DB))
`resource_bounds`	`ResourceBounds`	field(default_factory=ResourceBounds)
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`infer(data: MCPOpenclawInput) -> Result[MCPOpenclawOutput]`¶

`MCPOpenclawRecord(BaseModel)`¶

A single metadata record returned from OpenclawStore queries.

Field	Type	Default
`id`	`str`	required
`record_type`	`str`	required
`name`	`str`	required
`content`	`str`	required
`tags`	`str`	''
`timestamp`	`str`	''
`metadata`	`dict`	Field(default_factory=dict)

`MCPOpenclawInput(BaseModel)`¶

Input to AgentOpenclawMCPBlock — 26 ops.

Field	Type	Default
`op`	`Literal['connect_channel', 'disconnect_channel', 'list_channels', 'channel_status', 'configure_channel', 'send_message', 'receive_message', 'broadcast', 'route_message', 'get_history', 'create_session', 'list_sessions', 'get_session', 'close_session', 'schedule_message', 'list_scheduled', 'cancel_scheduled', 'webhook_create', 'webhook_list', 'webhook_delete', 'store_contact', 'list_contacts', 'search_contacts', 'search', 'info', 'capabilities', 'list_patterns', 'ops', 'help']`	required
`channel_type`	`str`	''
`channel_id`	`str`	''
`message`	`str`	''
`target`	`str`	''
`config_json`	`str`	'{}'
`session_id`	`str`	''
`name`	`str`	''
`tags_csv`	`str`	''
`webhook_url`	`str`	''
`event_type`	`str`	''
`contact_name`	`str`	''
`contact_data`	`str`	'{}'
`webhook_secret`	`str`	''
`schedule_time`	`str`	''
`schedule_id`	`str`	''
`query`	`str`	''
`top_k`	`int`	10
`limit`	`int`	50

`MCPOpenclawOutput(BaseModel)`¶

Output from AgentOpenclawMCPBlock.

Field	Type	Default
`op`	`str`	required
`success`	`bool`	False
`message`	`str`	''
`data`	`dict[str, Any]`	Field(default_factory=dict)
`records`	`list[dict]`	Field(default_factory=list)
`count`	`int`	0
`found`	`bool`	False
`retrieved`	`list[dict]`	Field(default_factory=list)
`scores`	`list[float]`	Field(default_factory=list)
`metadata`	`dict`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`str`	'qualified-draft'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`mode`	`str`	'control_plane'
`backend`	`str`	''
`runtime_available`	`bool`	False
`runtime_result`	`dict[str, Any]`	Field(default_factory=dict)
`audit_id`	`str`	''
`request_id`	`str`	''
`run_id`	`str`	''

`OpenclawStore`¶

SQLite-backed store for the agent_openclaw MCP sub-package.

Constructor:

Parameter	Type	Default
`db_path`	`str`	':memory:'

Methods:

`store_channel(name: str, channel_type: str = '', config_json: str = '{}', status: str = 'active', channel_id: str = '') -> str`¶

Persist a channel. Returns the assigned id.

`get_channel(channel_id: str) -> dict | None`¶

Retrieve a channel by id. Returns None if not found.

`list_channels(limit: int = 50) -> list[dict]`¶

List all registered channels.

`delete_channel(channel_id: str) -> bool`¶

Delete a channel by id. Returns True if deleted.

`update_channel_config(channel_id: str, config_json: str) -> bool`¶

Update a channel's config. Returns True if updated.

`update_channel_status(channel_id: str, status: str) -> bool`¶

Update a channel's status. Returns True if updated.

`store_message(channel_id: str, content: str, direction: str = 'out', target: str = '', session_id: str = '') -> str`¶

Store a message. Returns the assigned id.

`get_messages(channel_id: str, limit: int = 50) -> list[dict]`¶

Get messages for a channel.

`list_messages_for_channel(channel_id: str, limit: int = 50) -> list[dict]`¶

Alias for get_messages.

`create_session(channel_id: str = '', metadata_json: str = '{}') -> str`¶

Create a new session. Returns the assigned id.

`get_session(session_id: str) -> dict | None`¶

Retrieve a session by id.

`list_sessions(channel_id: str = '', limit: int = 50) -> list[dict]`¶

List sessions, optionally filtered by channel_id.

`close_session(session_id: str) -> bool`¶

Close a session. Returns True if closed.

`schedule_message(channel_id: str, content: str, target: str = '', schedule_time: str = '') -> str`¶

Schedule a message. Returns the assigned id.

`list_scheduled(channel_id: str = '', limit: int = 50) -> list[dict]`¶

List scheduled messages, optionally filtered by channel_id.

`cancel_scheduled(schedule_id: str) -> bool`¶

Cancel a scheduled message. Returns True if cancelled.

`create_webhook(name: str, url: str, event_type: str = '') -> str`¶

Create a webhook. Returns the assigned id.

`list_webhooks(limit: int = 50) -> list[dict]`¶

List all webhooks.

`delete_webhook(webhook_id: str) -> bool`¶

Delete a webhook by id. Returns True if deleted.

`store_contact(name: str, data_json: str = '{}', tags: str = '') -> str`¶

Store a contact. Returns the assigned id.

`list_contacts(tags_csv: str = '', limit: int = 50) -> list[dict]`¶

List contacts, optionally filtered by tags.

`search_contacts(query: str, top_k: int = 10) -> list[dict]`¶

TF-IDF search across contacts + channels (substring fallback).

`log_error(op: str, error_message: str, params: dict | None = None) -> None`¶

`record_audit_event(component: str, operation: str, action: str, params_summary: dict | None = None, backend: str = '', mode: str = 'control_plane', runtime_available: bool = False, result_status: str = '', error_text: str = '') -> str`¶

`count_all() -> dict[str, int]`¶

`text_search(query: str, top_k: int = 10) -> list[dict]`¶

Search contacts + channels by TF-IDF (falls back to substring).

MCP Tools¶

Operation	Source
`connect_channel`	openclaw_mcp
`disconnect_channel`	openclaw_mcp
`list_channels`	openclaw_mcp
`channel_status`	openclaw_mcp
`configure_channel`	openclaw_mcp
`send_message`	openclaw_mcp
`receive_message`	openclaw_mcp
`broadcast`	openclaw_mcp
`route_message`	openclaw_mcp
`get_history`	openclaw_mcp
`create_session`	openclaw_mcp
`list_sessions`	openclaw_mcp
`get_session`	openclaw_mcp
`close_session`	openclaw_mcp
`schedule_message`	openclaw_mcp
`list_scheduled`	openclaw_mcp
`cancel_scheduled`	openclaw_mcp
`webhook_create`	openclaw_mcp
`webhook_list`	openclaw_mcp
`webhook_delete`	openclaw_mcp
`store_contact`	openclaw_mcp
`list_contacts`	openclaw_mcp
`search_contacts`	openclaw_mcp
`search`	openclaw_mcp
`info`	openclaw_mcp
`capabilities`	openclaw_mcp
`list_patterns`	openclaw_mcp
`ops`	openclaw_mcp
`help`	openclaw_mcp

Agent Openclaw¶

Overview¶

Public API¶

AgentOpenclawBlock(LifecycleMixin, AIBlock[ChannelInput, ChannelOutput, None])¶

stop() -> Result[None]¶

infer(data: ChannelInput) -> Result[ChannelOutput]¶

ChannelInput(BaseModel)¶

ChannelOutput(BaseModel)¶

AgentOpenclawMCPBlock(AIBlock[MCPOpenclawInput, MCPOpenclawOutput, dict])¶

infer(data: MCPOpenclawInput) -> Result[MCPOpenclawOutput]¶

MCPOpenclawRecord(BaseModel)¶

MCPOpenclawInput(BaseModel)¶

MCPOpenclawOutput(BaseModel)¶

OpenclawStore¶

store_channel(name: str, channel_type: str = '', config_json: str = '{}', status: str = 'active', channel_id: str = '') -> str¶

get_channel(channel_id: str) -> dict | None¶

list_channels(limit: int = 50) -> list[dict]¶

delete_channel(channel_id: str) -> bool¶

update_channel_config(channel_id: str, config_json: str) -> bool¶

update_channel_status(channel_id: str, status: str) -> bool¶

store_message(channel_id: str, content: str, direction: str = 'out', target: str = '', session_id: str = '') -> str¶

get_messages(channel_id: str, limit: int = 50) -> list[dict]¶

list_messages_for_channel(channel_id: str, limit: int = 50) -> list[dict]¶

create_session(channel_id: str = '', metadata_json: str = '{}') -> str¶

get_session(session_id: str) -> dict | None¶

list_sessions(channel_id: str = '', limit: int = 50) -> list[dict]¶

close_session(session_id: str) -> bool¶

schedule_message(channel_id: str, content: str, target: str = '', schedule_time: str = '') -> str¶

list_scheduled(channel_id: str = '', limit: int = 50) -> list[dict]¶

cancel_scheduled(schedule_id: str) -> bool¶

create_webhook(name: str, url: str, event_type: str = '') -> str¶

list_webhooks(limit: int = 50) -> list[dict]¶

delete_webhook(webhook_id: str) -> bool¶

store_contact(name: str, data_json: str = '{}', tags: str = '') -> str¶

list_contacts(tags_csv: str = '', limit: int = 50) -> list[dict]¶

search_contacts(query: str, top_k: int = 10) -> list[dict]¶

log_error(op: str, error_message: str, params: dict | None = None) -> None¶

record_audit_event(component: str, operation: str, action: str, params_summary: dict | None = None, backend: str = '', mode: str = 'control_plane', runtime_available: bool = False, result_status: str = '', error_text: str = '') -> str¶

count_all() -> dict[str, int]¶

text_search(query: str, top_k: int = 10) -> list[dict]¶

MCP Tools¶

`AgentOpenclawBlock(LifecycleMixin, AIBlock[ChannelInput, ChannelOutput, None])`¶

`stop() -> Result[None]`¶

`infer(data: ChannelInput) -> Result[ChannelOutput]`¶

`ChannelInput(BaseModel)`¶

`ChannelOutput(BaseModel)`¶

`AgentOpenclawMCPBlock(AIBlock[MCPOpenclawInput, MCPOpenclawOutput, dict])`¶

`infer(data: MCPOpenclawInput) -> Result[MCPOpenclawOutput]`¶

`MCPOpenclawRecord(BaseModel)`¶

`MCPOpenclawInput(BaseModel)`¶

`MCPOpenclawOutput(BaseModel)`¶

`OpenclawStore`¶

`store_channel(name: str, channel_type: str = '', config_json: str = '{}', status: str = 'active', channel_id: str = '') -> str`¶

`get_channel(channel_id: str) -> dict | None`¶

`list_channels(limit: int = 50) -> list[dict]`¶

`delete_channel(channel_id: str) -> bool`¶

`update_channel_config(channel_id: str, config_json: str) -> bool`¶

`update_channel_status(channel_id: str, status: str) -> bool`¶

`store_message(channel_id: str, content: str, direction: str = 'out', target: str = '', session_id: str = '') -> str`¶

`get_messages(channel_id: str, limit: int = 50) -> list[dict]`¶

`list_messages_for_channel(channel_id: str, limit: int = 50) -> list[dict]`¶

`create_session(channel_id: str = '', metadata_json: str = '{}') -> str`¶

`get_session(session_id: str) -> dict | None`¶

`list_sessions(channel_id: str = '', limit: int = 50) -> list[dict]`¶

`close_session(session_id: str) -> bool`¶

`schedule_message(channel_id: str, content: str, target: str = '', schedule_time: str = '') -> str`¶

`list_scheduled(channel_id: str = '', limit: int = 50) -> list[dict]`¶

`cancel_scheduled(schedule_id: str) -> bool`¶

`create_webhook(name: str, url: str, event_type: str = '') -> str`¶

`list_webhooks(limit: int = 50) -> list[dict]`¶

`delete_webhook(webhook_id: str) -> bool`¶

`store_contact(name: str, data_json: str = '{}', tags: str = '') -> str`¶

`list_contacts(tags_csv: str = '', limit: int = 50) -> list[dict]`¶

`search_contacts(query: str, top_k: int = 10) -> list[dict]`¶

`log_error(op: str, error_message: str, params: dict | None = None) -> None`¶

`record_audit_event(component: str, operation: str, action: str, params_summary: dict | None = None, backend: str = '', mode: str = 'control_plane', runtime_available: bool = False, result_status: str = '', error_text: str = '') -> str`¶

`count_all() -> dict[str, int]`¶

`text_search(query: str, top_k: int = 10) -> list[dict]`¶