From oo
This skill is the front door for everything `oo`. Its job is to detect what the user wants and either **route to the right sub-skill** or **handle the connect flow inline**.
How this skill is triggered — by the user, by Claude, or both
Slash command
/oo:ooThe summary Claude sees in its skill listing — used to decide when to auto-load this skill
This skill is the front door for everything `oo`. Its job is to detect what the user wants and either **route to the right sub-skill** or **handle the connect flow inline**.
oo OrchestratorThis skill is the front door for everything oo. Its job is to detect what the user wants and either route to the right sub-skill or handle the connect flow inline.
Read the user's message and classify into ONE of:
| Intent | Signals | Action |
|---|---|---|
| connect | message contains 0x[0-9a-fA-F]{64}, or words like "connect", "delegate", "ask agent", "send to agent" | Continue with the Connect flow below |
| init | "init", "set up oo", "scaffold", "create agent.json", "make me publishable", "start a new bundle" | Invoke the oo-init skill via the Skill tool |
| publish | "publish", "ship", "share my agent", "release my skills", "announce my agent" | Invoke oo-publish |
| subscribe | "subscribe", "follow", "install 's skills", "add 0x... as a subscription" | Invoke oo-subscribe |
| accept | "accept subscribers", "review subscription requests", "who's following me", "approve subscriptions" | Invoke oo-accept |
| ambiguous | bare /oo, "oo", "help me with oo", or anything that fits >1 bucket | Go to Step 2 |
If the intent is clear, route immediately. Don't ask.
If you cannot confidently classify, ask the user once with AskUserQuestion. Tailor the options based on whether they already have an identity:
~/.co/agent.json does not exist → likely init is needed first; offer init / subscribe / connect.Then invoke the chosen sub-skill. Never proceed past Step 2 without a clear intent.
The rest of this file handles the connect intent only. Everything else is delegated to its own skill.
Connect to remote ConnectOnion agents, delegate tasks, and handle multi-turn collaboration.
Before any interaction, verify the environment is ready. Run these checks sequentially — stop on first failure:
1. Check connectonion is installed:
python -c "import connectonion; print(connectonion.__version__)"
If ImportError: run pip install connectonion, then re-check.
2. Check agent identity exists:
ls .co/keys/agent.key 2>/dev/null || ls ~/.co/keys/agent.key 2>/dev/null
If neither exists: run co init to generate identity.
3. Verify identity is usable:
python -c "
from connectonion import address
from pathlib import Path
a = address.load(Path('.co')) or address.load(Path.home() / '.co')
print(a['address'])
"
If fails: report the error and stop. The user needs to fix their .co/ directory.
Note:
import connectonionprints[env] ...lines to stdout. For all environment checks, parse only the last line of stdout output. Ignore everything else.
Skip environment checks after the first successful run in a session.
Extract from the user's message:
0x[0-9a-fA-F]{64} (66 chars total)For /oo slash command: /oo <address> <task description>
The default connect() library has a known issue: the relay API may not return an online field, causing direct endpoint resolution to always fail and falling back to relay — which itself may be unreliable. To work around this, the skill uses a smart connection script that:
/info)Generate and execute this Python script (fill in {address} and {task}):
import sys, json, time, uuid, asyncio
import httpx, websockets
from connectonion import address
from pathlib import Path
TARGET = "{address}"
TASK = "{task}"
TIMEOUT = 60
RELAY_URL = "wss://oo.openonion.ai"
keys = address.load(Path(".co")) or address.load(Path.home() / ".co")
def _sort_endpoints(endpoints):
def priority(url):
if "localhost" in url or "127.0.0.1" in url:
return 0
if any(x in url for x in ("192.168.", "10.", "172.16.", "172.17.", "172.18.")):
return 1
return 2
return sorted(endpoints, key=priority)
def discover_direct_ws(target, relay_url):
"""Query relay API for endpoints and find a working direct WebSocket."""
https_relay = relay_url.replace("wss://", "https://").replace("ws://", "http://").rstrip("/")
try:
resp = httpx.get(f"{https_relay}/api/relay/agents/{target}", timeout=5)
if resp.status_code != 200:
return None
info = resp.json()
except Exception:
return None
endpoints = info.get("endpoints", [])
if not endpoints:
return None
http_endpoints = [ep for ep in _sort_endpoints(endpoints)
if ep.startswith("http://") or ep.startswith("https://")]
for http_url in http_endpoints:
try:
r = httpx.get(f"{http_url}/info", timeout=3, proxy=None)
if r.status_code == 200 and r.json().get("address") == target:
ws_url = http_url.replace("https://", "wss://").replace("http://", "ws://")
if not ws_url.endswith("/ws"):
ws_url = ws_url.rstrip("/") + "/ws"
return ws_url
except Exception:
continue
return None
async def direct_connect(ws_url, target, keys, task, timeout):
"""Connect directly to agent WebSocket, send task, return result."""
async with websockets.connect(ws_url, proxy=None) as ws:
# Signed CONNECT
ts = int(time.time())
payload = {"to": target, "timestamp": ts}
canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
signature = address.sign(keys, canonical.encode())
connect_msg = {
"type": "CONNECT", "timestamp": ts, "to": target,
"payload": payload, "from": keys["address"], "signature": signature.hex()
}
await ws.send(json.dumps(connect_msg))
# Wait for CONNECTED
raw = await asyncio.wait_for(ws.recv(), timeout=10)
event = json.loads(raw)
if event.get("type") == "ERROR":
raise ConnectionError(f"Auth error: {event.get('message', event.get('error'))}")
if event.get("type") != "CONNECTED":
raise ConnectionError(f"Unexpected: {event.get('type')}")
# Signed INPUT
ts2 = int(time.time())
input_id = str(uuid.uuid4())
input_payload = {"prompt": task, "timestamp": ts2}
input_canonical = json.dumps(input_payload, sort_keys=True, separators=(",", ":"))
input_sig = address.sign(keys, input_canonical.encode())
input_msg = {
"type": "INPUT", "input_id": input_id, "prompt": task, "timestamp": ts2,
"payload": input_payload, "from": keys["address"], "signature": input_sig.hex()
}
await ws.send(json.dumps(input_msg))
# Stream until OUTPUT
while True:
msg = await asyncio.wait_for(ws.recv(), timeout=timeout)
ev = json.loads(msg)
t = ev.get("type")
if t == "OUTPUT":
return ev.get("result", ""), True
elif t == "ask_user":
return ev.get("text", ""), False
elif t == "ERROR":
raise ConnectionError(f"Agent error: {ev.get('message', ev.get('error'))}")
# --- Main ---
result_text, done = None, None
# Step 1: Try direct connection
ws_url = discover_direct_ws(TARGET, RELAY_URL)
if ws_url:
try:
result_text, done = asyncio.run(direct_connect(ws_url, TARGET, keys, TASK, TIMEOUT))
print(f"CO_METHOD: direct", flush=True)
except Exception as e:
print(f"CO_DIRECT_FAIL: {e}", flush=True)
# Step 2: Fallback to relay
if result_text is None:
try:
from connectonion import connect
agent = connect(TARGET, keys=keys)
response = agent.input(TASK, timeout=TIMEOUT)
result_text, done = response.text, response.done
print(f"CO_METHOD: relay", flush=True)
except Exception as e:
print(f"CO_RELAY_FAIL: {e}", flush=True)
sys.exit(1)
print(f"CO_RESPONSE: {json.dumps(result_text)}", flush=True)
print(f"CO_DONE: {done}", flush=True)
Execute via your shell tool. Parse stdout — only lines starting with CO_ matter, ignore all others. The CO_RESPONSE value is JSON-encoded (to handle multi-line responses). Decode it before presenting to the user.
CO_DONE: True → return CO_RESPONSE content to the user. Done.CO_DONE: False → the remote agent is asking a follow-up question. See Multi-turn Task below.CO_METHOD: direct → connected directly (fastest path).CO_METHOD: relay → connected via relay fallback.CO_DIRECT_FAIL: ... → direct failed, trying relay next.CO_RELAY_FAIL: ... → both methods failed. Report error to user.For long-running tasks, increase timeout to 300.
Multi-turn requires maintaining session state. Use separate one-shot calls per turn since stdin interaction is unreliable in agent environments. Each turn is a new connection but passes context through the conversation.
For the first turn, use the one-shot script above. For follow-up turns, include conversation context in the prompt (e.g., prepend prior exchanges).
After each round, parse stdout — only CO_ prefixed lines matter, ignore all others:
CO_DONE: True → return CO_RESPONSE content to the user. Done.CO_DONE: False → the remote agent asked a follow-up question:
CO_RESPONSE to the user, wait for their reply, then send another round.CO_DONE: True or 10 rounds.If the script fails, check stderr for these patterns:
| Error in stderr | Cause | Action |
|---|---|---|
ImportError: No module named 'connectonion' | Not installed | Run pip install connectonion |
address.load() returns None | No identity | Run co init |
TimeoutError | Remote agent unreachable or slow | Verify address, check network/proxy, increase timeout |
ConnectionRefused or relay lookup fail | Agent offline | Confirm remote agent is running with host() |
CO_DIRECT_FAIL + CO_RELAY_FAIL | Both paths failed | Agent likely offline — check with operator |
| Trust/permission error | Not authorized | Tell user to contact the remote agent admin for access |
InsufficientCreditsError | No credits for co/ models | Run co status to check balance |
| Script hangs (no output for >90s) | Remote agent requesting onboard (invite code/payment) | Kill script, tell user the remote agent requires onboarding |
Provides behavioral guidelines to reduce common LLM coding mistakes, focusing on simplicity, surgical changes, assumption surfacing, and verifiable success criteria.
Searches, retrieves, and installs Agent Skills from prompts.chat registry using MCP tools like search_skills and get_skill. Activates for finding skills, browsing catalogs, or extending Claude.
Creates, edits, and optimizes skills for Claude Code, including drafting, evaluating with test prompts, iterating on performance, and improving skill descriptions for better triggering accuracy.
npx claudepluginhub openonion/oo --plugin oo