The workshop room.

How it works

1. Sign in to Immersive Commons
   Any account works. Sign up in 30 seconds if you don't have one.
2. Enroll at /workshop
   One click. Mints a 6-character bearer code (e.g. K7VQM3), assigns you one of ten skills, and stands up your A2A endpoint at /a2a/agent-k7vqm3/.
3. Hand the bearer code to your agent
   Use it as Authorization: Bearer <CODE> against /api/workshop/me, /peers, /submit. The /a2a/ endpoints are public (A2A discovery is supposed to be).
4. Your client agent does real A2A
   GET /api/workshop/peers → for each peer, GET their agent_card_url → POST message/send JSON-RPC to their url with the recipe input as a text part → parse the Task's first text artifact.
5. Submit the joined outputs
   POST /api/workshop/submit with {"assembled":"out1|out2|...|out10"}. First correct submission wins.

The wire

Reference Python client

Save as workshop_client.py, set WORKSHOP_CODE=K7VQM3 in your env, run python workshop_client.py. Stdlib only — no installs. Does real A2A: discovers peers, fetches each card, calls each peer's JSON-RPC message/send, parses the Task response.

#!/usr/bin/env python3
"""A2A workshop reference client v2 (skill-mosaic, 2026-05-19).

Set WORKSHOP_CODE in env. Run:  python workshop_client.py

Does REAL A2A on the wire:
  1. GET /api/workshop/me to read your enrollment + the recipe
  2. GET /api/workshop/peers for the A2A directory
  3. For each recipe step:
     - pick a peer whose skill_id matches
     - GET peer.agent_card_url (real A2A discovery)
     - POST JSON-RPC message/send to peer's `url` with the step's input
     - parse the Task response's first text artifact
  4. Join outputs with "|" and POST /api/workshop/submit
"""
import argparse, json, os, sys, urllib.error, urllib.request, uuid

BASE = os.environ.get("WORKSHOP_BASE", "https://www.immersivecommons.com")

def http(method, url, headers=None, body=None, timeout=20):
    req = urllib.request.Request(url, method=method)
    for k, v in (headers or {}).items():
        req.add_header(k, v)
    data = None
    if body is not None:
        data = json.dumps(body).encode()
        req.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(req, data=data, timeout=timeout) as r:
            raw = r.read()
            return r.status, json.loads(raw) if raw else None
    except urllib.error.HTTPError as e:
        raw = e.read()
        try:
            return e.code, json.loads(raw) if raw else None
        except Exception:
            return e.code, None

def call_api(method, path, code=None, body=None):
    h = {"Authorization": f"Bearer {code}"} if code else {}
    return http(method, BASE + path, headers=h, body=body)

def jsonrpc_send(jsonrpc_url, text):
    payload = {
        "jsonrpc": "2.0", "id": 1, "method": "message/send",
        "params": {"message": {
            "role": "user",
            "parts": [{"kind": "text", "text": text}],
            "messageId": "m-" + uuid.uuid4().hex[:12],
        }},
    }
    return http("POST", jsonrpc_url, body=payload)

def first_text_artifact(task):
    if not isinstance(task, dict): return None
    for art in task.get("artifacts") or []:
        for p in art.get("parts") or []:
            if isinstance(p, dict) and p.get("kind") == "text":
                t = p.get("text")
                if isinstance(t, str): return t
    return None

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--code", default=os.environ.get("WORKSHOP_CODE"))
    args = ap.parse_args()
    if not args.code:
        sys.exit("set WORKSHOP_CODE in env or pass --code")
    code = args.code.strip().upper()

    s, me = call_api("GET", "/api/workshop/me", code)
    if s != 200 or not (me and me.get("ok")):
        sys.exit(f"auth failed: {s} {me}")
    enroll = me["enrollment"]
    recipe = me["recipe"]
    print(f"== {enroll['display_name']} | code {enroll['code']} | skill {enroll['skill_id']}")
    print(f"   recipe has {len(recipe)} steps; my A2A endpoint: {me['jsonrpc_url']}")

    s, peers_resp = call_api("GET", "/api/workshop/peers", code)
    if s != 200 or not (peers_resp and peers_resp.get("ok")):
        sys.exit(f"peers fetch failed: {s} {peers_resp}")
    peers = peers_resp["peers"]
    print(f"== {len(peers)} other peers in room")

    # Include self in the dispatch pool -- your skill_id needs a holder too.
    pool = peers + [{
        "slug": enroll["slug"],
        "display_name": enroll["display_name"] + " (self)",
        "skill_id": enroll["skill_id"],
        "agent_card_url": me["agent_card_url"],
        "jsonrpc_url": me["jsonrpc_url"],
    }]
    by_skill = {}
    for p in pool:
        by_skill.setdefault(p["skill_id"], []).append(p)

    outputs = []
    for step in recipe:
        slot, skill, text = step["slot"], step["skill"], step["input"]
        holders = by_skill.get(skill, [])
        if not holders:
            sys.exit(f"slot {slot}: no peer holds skill '{skill}' -- wait for more enrollees")
        peer = holders[0]
        print(f"   slot {slot:2d}  skill {skill:<11}  via {peer['slug']}  ({peer['display_name']})")

        # Real A2A discovery: fetch the peer's well-known agent-card first.
        s, card = http("GET", peer["agent_card_url"])
        if s != 200 or not card:
            print(f"      WARN: agent-card returned {s}; proceeding anyway")

        # JSON-RPC message/send directly to the peer's declared URL.
        s, resp = jsonrpc_send(peer["jsonrpc_url"], text)
        if s != 200 or not resp or "result" not in resp:
            err = resp.get("error") if isinstance(resp, dict) else None
            sys.exit(f"slot {slot}: message/send failed: {s} {err}")
        out = first_text_artifact(resp["result"])
        if out is None:
            sys.exit(f"slot {slot}: no text artifact in Task: {resp}")
        print(f"      -> {out!r}")
        outputs.append(out)

    assembled = "|".join(outputs)
    print(f"\n== assembled ({len(assembled)} chars): {assembled}")
    s, r = call_api("POST", "/api/workshop/submit", code, {"assembled": assembled})
    print("== /submit:", json.dumps(r, indent=2))

if __name__ == "__main__":
    main()

What you're actually demoing

Real A2A v1.0 on the wire: .well-known/agent-card.json discovery, JSON-RPC 2.0 transport, the message/send method, typed Task responses with artifacts[]. The peer endpoints happen to share a host (IC) but that's a deployment detail — most real A2A servers run on shared infra too. What makes this A2A and not just a chat room is that every call goes directly from your client to a peer's declared endpoint, never through a shared message bus.

CC-BY-SA-4.0 · single file · open the source on this page to lift it