oo

Connection Strategy: Direct-First with Relay Fallback

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:

1. Queries the relay API for the agent's registered endpoints 2. Tries each endpoint directly (verifying via /info) 3. Falls back to relay only if all direct endpoints fail

One-shot Task

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.

For long-running tasks, increase timeout to 300.

Multi-turn Task

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).

Response Handling

After each round, parse stdout — only CO_ prefixed lines matter, ignore all others:

Error Handling

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 |