Skip to content

EMM A2A Phase 2: Task Manager (list_board)

Українська версія

EMM A2A Phase 2 Routing

Context: Multi-Agent Agent Card

In A2A one Agent Card can describe multiple skills from different agents. It's not "one endpoint = one agent" — it's "one endpoint = one A2A server" that routes by skillId to the right LangGraph. Client gets one JSON with all skills and decides which to call. No need for separate URLs per agent.

Phase 1 had only Process Manager. In Phase 2 I added Task Manager. Same POST /api/a2a/tasks, but now Adapter looks at params.skillId: if list_board — calls Task Manager; if missing or weekly_report — Process Manager. Default fallback — Process Manager, so legacy clients without skillId keep working.

Routing Diagram

flowchart LR
    subgraph Client[A2A Client]
        R[Request]
    end

    subgraph Router[A2A Router]
        P[JSON-RPC Parser]
        ROUTE{skillId?}
        PM_MAP[ProcessManagerMapper]
        TM_MAP[TaskManagerMapper]
    end

    subgraph Agents[Agents]
        PM[Process Manager\nweekly_report]
        TM[Task Manager\nlist_board]
    end

    R --> P
    P --> ROUTE
    ROUTE -->|"default / weekly_report"| PM_MAP
    ROUTE -->|"list_board"| TM_MAP
    PM_MAP --> PM
    TM_MAP --> TM

Sequence: list_board

sequenceDiagram
    participant C as A2A Client
    participant API as FastAPI
    participant Adapter as A2A Adapter
    participant TM as Task Manager
    participant Store as Task Store / lakeFS

    C->>API: GET /.well-known/agent.json
    API->>C: Agent Card (skills: weekly_report, list_board)

    C->>API: POST /api/a2a/tasks
    Note over C,API: skillId: list_board, message.parts[0].text: {"operation":"list_board","board_id":"..."}

    API->>Adapter: parse, skillId = list_board
    Adapter->>Adapter: TaskManagerMapper: A2A → LangGraph input

    Adapter->>TM: graph.ainvoke({"operation":"list_board","board_id":"..."})
    TM->>Store: read boards, lists, cards
    Store-->>TM: board JSON
    TM-->>Adapter: result (board structure)

    Adapter->>Adapter: map → A2A Task + Artifact
    Adapter-->>C: JSON-RPC {result: {task: {artifact: board}}}

What I Added

  • skillId — optional param. Missing or weekly_report → Process Manager. list_board → Task Manager. Unknown skillId — currently fallback to Process Manager (may fail if operation not found; explicit validation later).

  • TaskManagerMapper — same idea as ProcessManagerMapper, but for Task Manager. Input: {"operation":"list_board","board_id":"..."}. Output — artifact with board JSON: structure with lists, each list has cards with id, title, description. Same format as Task Manager API so client can render or process immediately.

  • Agent Card — now two skills. Client sees both and chooses. Discovery once — then calls with needed skillId.

Code: Routing by skill_id in _handle_tasks_submit. Default — weekly_report (Process Manager). list_board → Task Manager. TaskManagerMapper — same pattern as ProcessManagerMapper: a2a_message_to_input parses JSON from message.parts, result_to_a2a_task wraps into A2A Task.

# viz/backend/viz_backend/routers/a2a.py
async def _handle_tasks_submit(params: dict, req_id) -> dict:
    skill_id = params.get("skillId") or "weekly_report"
    if skill_id == "weekly_report":
        return await _handle_process_manager(params, req_id)
    if skill_id == "list_board":
        return await _handle_task_manager(params, req_id)
    # unknown skillId → failed task

# viz/backend/viz_backend/a2a/task_manager_mapper.py
def a2a_message_to_input(message: dict) -> dict:
    merged = {"operation": "list_board", "board_id": None}
    for t in text_parts:
        data = json.loads(t) if t.startswith("{") else {"board_id": t}
        for k, v in data.items():
            if v is not None and k in merged:
                merged[k] = v
    return merged

Example

curl -X POST http://localhost:8000/api/a2a/tasks \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "tasks/submit",
    "params": {
      "taskId": "test-2",
      "skillId": "list_board",
      "message": {
        "role": "user",
        "parts": [{
          "type": "text",
          "text": "{\"operation\":\"list_board\",\"board_id\":\"<BOARD_ID>\"}"
        }]
      }
    },
    "id": 2
  }'

Replace <BOARD_ID> with a real board ID from Task Manager. Get it from frontend (URL when opening a board) or via list_workspaces → list_boards in MCP Task Manager. Response contains full board structure: lists with cards, ready to parse for further processing.

Limitations

  • Only list_board so far. Create card, move card, list workspaces — not in A2A yet. Read-only. Write operations need more params (list_id, card title, position) — I'll add as separate skill.
  • Invalid board_id → status: "failed" with error text in artifact. Task Manager returns error — Adapter passes it in A2A format.

Why I Chose list_board

Simplest operation: one input (board_id), one output (board JSON). Other Task Manager operations have more params: create_card needs list_id, title; move_card — card_id, target_list_id. list_board — minimal pilot for second agent: verify routing and mapping without complex schemas.

What's Next

Phase 2+ — TaskStore and tasks/status. To get taskId and poll status later, without blocking on submit. Then Phase 3 — streaming via SSE.