Skip to content

Agent Nanoclaw

Agent NanoClaw — mvp.agent_nanoclaw

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

Overview

Group messaging block with SQLite WAL-mode persistence for multi-agent coordination. Provides 8 operations — create/list/get/delete groups and send/receive/broadcast messages — with durable storage at a configurable DB path. Enforces TOS constraints by blocking Anthropic and Claude Code as providers; only ollama and openrouter are permitted.

Control-plane first

agent_nanoclaw is currently a local control-plane and persistence layer. Group, message, task, memory, skill, sandbox, and audit records are stored in SQLite, but task scheduling, sandbox isolation, and skill invocation do not by themselves execute user workflows unless a real NanoClaw runtime backend is configured. Treat results with degraded=True, mode="control_plane", or mode="simulated" as recorded metadata rather than completed runtime work.

Launch positioning

For first-user or pilot demos, position NanoClaw as a coordination ledger and audit surface. Do not present schedule_task as a background job runner, create_sandbox as process/container isolation, or invoke_skill as real skill execution unless the deployment has an available NanoClaw backend and the returned runtime metadata confirms it.

When to use:

  • Coordinating messages between G6 agents via named groups without a full message broker
  • Persisting inter-agent communication across process restarts with WAL-safe SQLite
  • Building group-chat primitives that satisfy open-source provider TOS requirements
  • Recording task, memory, sandbox, skill, and audit state for workflows that are executed elsewhere

Example:

from mvp.agent_nanoclaw import AgentNanoclawBlock, GroupInput

block = AgentNanoclawBlock(name="nanoclaw")
block.infer(GroupInput(op="create_group", group_id="team-alpha", group_name="Alpha"))
result = block.infer(GroupInput(
    op="send", group_id="team-alpha",
    message="Task ready", agent_id="solver", provider="ollama",
))
# result.ok → True; result.value.success → True

Works well with: agent_openclaw, autonomous_orchestrator, hat_orchestrator

Public API

AgentNanoclawBlock(LifecycleMixin, AIBlock[GroupInput, GroupOutput, None])

Group messaging store with SQLite WAL persistence.

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

Methods:

stop() -> Result[None]

infer(data: GroupInput) -> Result[GroupOutput]

GroupInput(BaseModel)

Input to AgentNanoclawBlock — 8 basic ops.

Field Type Default
op Literal['create_group', 'list_groups', 'send', 'receive', 'broadcast', 'get_group', 'configure', 'delete_group'] required
group_id str ''
group_name str ''
message str ''
target str ''
config dict[str, Any] Field(default_factory=dict)
agent_id str ''
provider str 'ollama'

GroupOutput(BaseModel)

Output from AgentNanoclawBlock.

Field Type Default
op str required
group_id str ''
success bool False
message str ''
data dict[str, Any] Field(default_factory=dict)
groups list[dict[str, Any]] Field(default_factory=list)
degraded bool False
degradation_reason str ''
completion_state str 'qualified-draft'
warning_card str ''
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 ''

AgentNanoclawMCPBlock(AIBlock[MCPNanoclawInput, MCPNanoclawOutput, dict])

28-op NanoClaw MCP block with SQLite persistence (25 + capabilities + advisory ops).

Field Type Default
name str 'agent_nanoclaw_mcp'
state dict field(default_factory=dict)
db_path str field(default_factory=lambda: os.environ.get('NANOCLAW_DB_PATH', _DEFAULT_DB))
resource_bounds ResourceBounds field(default_factory=ResourceBounds)
usage ResourceUsage field(default_factory=ResourceUsage)
agentic_planner object None

Methods:

infer(data: MCPNanoclawInput) -> Result[MCPNanoclawOutput]

MCPNanoclawRecord(BaseModel)

A single metadata record returned from NanoclawStore 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)

MCPNanoclawInput(BaseModel)

Input to AgentNanoclawMCPBlock — 28 ops.

Field Type Default
op Literal['create_group', 'list_groups', 'get_group', 'configure_group', 'delete_group', 'send_message', 'receive_message', 'broadcast', 'route_message', 'get_history', 'schedule_task', 'list_tasks', 'get_task', 'cancel_task', 'create_sandbox', 'sandbox_status', 'destroy_sandbox', 'store_memory', 'retrieve_memory', 'list_memories', 'register_skill', 'list_skills', 'invoke_skill', 'search', 'info', 'capabilities', 'recommend_backend', 'list_patterns'] required
group_id str ''
group_name str ''
message str ''
target str ''
config_json str '{}'
agent_id str ''
name str ''
tags_csv str ''
task_description str ''
task_id str ''
sandbox_id str ''
sandbox_config str '{}'
memory_key str ''
memory_value str ''
skill_name str ''
skill_args str '{}'
query str ''
top_k int 10
limit int 50
schedule_time str ''
candidate_providers list[str] Field(default_factory=list)

MCPNanoclawOutput(BaseModel)

Output from AgentNanoclawMCPBlock.

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[str, Any] Field(default_factory=dict)
degraded bool False
degradation_reason str ''
completion_state str 'qualified-draft'
warning_card str ''
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 ''

NanoclawStore

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

Constructor:

Parameter Type Default
db_path str ':memory:'

Methods:

create_group(name: str, config_json: str = '{}', agent_count: int = 0) -> str

Create a group. Returns the assigned id.

get_group(group_id: str) -> dict | None

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

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

List all groups.

configure_group(group_id: str, config_json: str) -> bool

Update group config. Returns True if group was found.

delete_group(group_id: str) -> bool

Delete a group. Returns True if group was found.

delete_group_cascade(group_id: str) -> dict[str, int]

Delete a group AND all related rows from messages, tasks,

store_message(group_id: str, sender: str, content: str, target: str = '') -> str

Store a message. Returns the assigned id.

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

Get messages for a group, most recent first.

list_messages_for_group(group_id: str, limit: int = 50) -> list[dict]

Alias for get_messages.

schedule_task(group_id: str, description: str, schedule_time: str = '') -> str

Schedule a task. Returns the assigned id.

get_task(task_id: str) -> dict | None

Retrieve a task by id.

list_tasks(group_id: str = '', limit: int = 50) -> list[dict]

List tasks, optionally filtered by group_id.

cancel_task(task_id: str) -> bool

Cancel a task. Returns True if task was found and not already cancelled.

create_sandbox(group_id: str = '', config_json: str = '{}') -> str

Create a sandbox. Returns the assigned id.

sandbox_status(sandbox_id: str) -> dict | None

Get sandbox status. Returns None if not found.

destroy_sandbox(sandbox_id: str) -> bool

Destroy (stop) a sandbox. Returns True if sandbox was found.

store_memory(group_id: str, key: str, value: str, tags: str = '') -> str

Store a memory entry. Returns the assigned id.

retrieve_memory(key: str, group_id: str = '') -> dict | None

Retrieve the latest memory entry by key (optionally scoped to group).

list_memories(group_id: str = '', limit: int = 50) -> list[dict]

List memory entries, optionally filtered by group_id.

register_skill(name: str, description: str = '', args_schema: str = '{}', group_id: str = '') -> str

Register a skill. Returns the assigned id.

list_skills(group_id: str = '', limit: int = 50) -> list[dict]

List registered skills, optionally filtered by group_id.

get_skill(name: str) -> dict | None

Retrieve a skill by name.

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 groups + memories + skills by TF-IDF (falls back to substring).

MCP Tools

Operation Source
create_group nanoclaw_mcp
list_groups nanoclaw_mcp
get_group nanoclaw_mcp
configure_group nanoclaw_mcp
delete_group nanoclaw_mcp
send_message nanoclaw_mcp
receive_message nanoclaw_mcp
broadcast nanoclaw_mcp
route_message nanoclaw_mcp
get_history nanoclaw_mcp
schedule_task nanoclaw_mcp
list_tasks nanoclaw_mcp
get_task nanoclaw_mcp
cancel_task nanoclaw_mcp
create_sandbox nanoclaw_mcp
sandbox_status nanoclaw_mcp
destroy_sandbox nanoclaw_mcp
store_memory nanoclaw_mcp
retrieve_memory nanoclaw_mcp
list_memories nanoclaw_mcp
register_skill nanoclaw_mcp
list_skills nanoclaw_mcp
invoke_skill nanoclaw_mcp
search nanoclaw_mcp
info nanoclaw_mcp
capabilities nanoclaw_mcp
recommend_backend nanoclaw_mcp
list_patterns nanoclaw_mcp

Production Caveats

  • Preserve runtime metadata in all MCP wrappers and UIs: degraded, degradation_reason, mode, backend, runtime_available, runtime_result, and audit_id.
  • Validate JSON strings before storing config_json, sandbox_config, or skill_args; invalid JSON may otherwise be accepted or silently treated as {}.
  • Do not rely on success=True alone for user-facing completion. For runtime-backed operations, also check runtime_available and mode.
  • Treat schedule_task as a pending task record, not a background execution guarantee.
  • Treat create_sandbox as a state record, not isolation.
  • Treat invoke_skill as simulated unless runtime metadata proves a real backend handled it.
  • Clean up related records explicitly when deleting a group if the workflow needs deletion semantics or privacy guarantees.