Acp Task Dispatch

ACP Task Dispatch

Send a discrete, self-contained task to the OpenClaw-mcode-ACP server instead of running it inline in the current session. Useful when:

• The task is long-running and you do not want to block
• You want a persistent record (SQLite history) for later review
• A worker pool should pick it up off the queue
• You want to observe progress via SSE / WebSocket events

Setup

The Plugin ships its own HTTP client. There is no ACP_HOME to set, no external Python SDK to install, and no sys.path to mutate. The client lives at <plugin_root>/client/_acp_client.py and is resolved through the ACP_PLUGIN_ROOT environment variable (set automatically by the Plugin runtime) with a __file__-based fallback for ad-hoc invocations.

Authentication

The bundled client reads the bearer token from one of (first hit wins):

1. $ACP_TOKEN (recommended for CI and shells)
2. ~/.acp_token (one line, no trailing newline)
3. <plugin_root>/.acp_token (one line; co-located fallback for fresh installs)

The client attaches Authorization: Bearer <token> to every request to http://127.0.0.1:9999/acp/*. Do not read, print, or pass the token yourself. The client also refuses to follow HTTP redirects (a hostile loopback server cannot exfiltrate the token via a 302) and refuses to talk to anything other than the loopback allow-list.

If the token cannot be located, the client raises ACPTokenMissing. Tell the user to set $ACP_TOKEN (or write one of the fallback files) and stop; do not retry.

Dispatch a task

import os, sys
# ACP_PLUGIN_ROOT is the directory that contains this Plugin's `client/`.
# It is set automatically when the Skill is loaded by the Plugin runtime;
# the `__file__` fallback keeps the snippet working when it is pasted
# into an ad-hoc Python session.
_plugin_root = os.environ.get('ACP_PLUGIN_ROOT') or os.path.dirname(
    os.path.dirname(os.path.abspath(__file__))
)
sys.path.insert(0, os.path.join(_plugin_root, 'client'))
from _acp_client import create_task, get_task, history

# create_task returns the task_id as a string directly (not a dict).
task_id = create_task(
    prompt="用一句话回答:1+1=?",
    workspace="D:/some/work/dir",
    timeout=300,
)
print(task_id)

create_task is fire-and-forget. The server runs the task on a worker pool (default 3 concurrent) and persists every transition to SQLite.

Poll for completion

import time
while True:
    state = get_task(task_id)
    # The terminal success state is `succeeded`, not `completed`.
    if state["status"] in ("succeeded", "failed", "timeout", "cancelled"):
        break
    time.sleep(2)
print(state.get("answer", state.get("error")))

For a blocking wait that returns the final task dict directly, use wait_task(task_id, timeout=600, poll_interval=2.0) from the same client.

Inspect history

# `history()` returns a list of task dicts directly, not
# `{"tasks": [...]}`.
for t in history(limit=20):
    print(t["task_id"], t["status"], t.get("duration_ms"))

Stream progress (optional)

stream_task(task_id, type, data: ...) consumes the server's SSE stream and yields {type, data} dicts. run_and_stream(prompt, workspace, ...) is a convenience that creates a task, streams its events, and returns the final task dict.

Constraints

• The prompt is the entire instruction given to a fresh mcode subprocess. It must be self-contained — the subprocess has no memory of your session.
• The workspace directory must exist; the server runs mcode with that as cwd.
• Default timeout is 60 seconds. Raise it for longer work, but consider --permission full first if the task needs to write files.
• For multi-step peer work, prefer the acp-collab Skill instead — this Skill is for one-shot fire-and-forget dispatch.

Failure handling

If create_task raises ACPError, the server is likely down or rejected the request. Verify the server is reachable on http://127.0.0.1:9999/acp/health and that the token matches. Stop and surface the error to the user; do not retry in a tight loop.