A2A · TASK LIFECYCLE

Task 생명주기

📐 상태 기계 다이어그램 📋 Task 객체 스펙 💻 상태 처리 코드

A2A Task의 상태 전이, Task 객체 구조, Artifact 타입, input-required 인터랙션 패턴을 완전히 이해합니다.

Task 상태 기계

A2A의 모든 작업은 Task 단위로 관리됩니다. Task는 생성 시점부터 종료까지 명확히 정의된 상태를 따라 전이합니다.

submitted working 추가 입력 필요 input-required 사용자 응답 completed 에러 발생 failed canceled completed / failed / canceled = 최종 상태 (변경 불가) 그림 1. A2A Task 상태 기계
상태의미다음 가능 상태
submittedTask가 서버에 접수됨. 아직 처리 시작 전working
working에이전트가 현재 처리 중completed, failed, canceled, input-required
input-required처리 중 사용자/클라이언트의 추가 입력 필요working (재개), canceled
completed작업 성공적으로 완료. Artifact 포함(최종)
failed에러로 작업 실패. 오류 메시지 포함(최종)
canceled클라이언트 또는 에이전트가 작업 취소(최종)

Task 객체 구조

json완료된 Task 객체 예시
{
  "id": "task-a1b2c3d4",
  "sessionId": "session-xyz",  // 다중 턴 대화 묶음
  "status": {
    "state": "completed",
    "message": {
      "role": "agent",
      "parts": [{ "type": "text", "text": "분석 완료되었습니다." }]
    },
    "timestamp": "2025-04-15T10:30:00Z"
  },
  "artifacts": [
    {
      "name": "analysis_result",
      "description": "분석 결과 JSON",
      "parts": [
        {
          "type": "data",
          "data": { "score": 0.92, "label": "positive" },
          "mimeType": "application/json"
        }
      ],
      "index": 0,
      "lastChunk": true
    }
  ],
  "history": [   // stateTransitionHistory: true 시 채워짐
    { "role": "user", "parts": [{ "type": "text", "text": "분석해줘" }] }
  ],
  "metadata": {}
}

Artifact Part 타입

type필드설명
texttext: string일반 텍스트, 마크다운
datadata: any, mimeType구조화된 JSON 데이터
filefile: FileContent인라인 파일 (base64) 또는 URI 참조

input-required 패턴 구현

에이전트가 처리 도중 사용자의 확인 또는 추가 정보가 필요할 때 input-required 상태로 전환합니다. 클라이언트는 같은 taskId와 sessionId로 추가 메시지를 보내 작업을 재개합니다.

pythoninput-required 처리 — 클라이언트 측
import httpx, uuid

async def run_task_with_followup(agent_url: str, message: str):
    session_id = str(uuid.uuid4())
    task_id = str(uuid.uuid4())

    payload = {
        "jsonrpc": "2.0", "id": 1,
        "method": "tasks/send",
        "params": {
            "id": task_id,
            "sessionId": session_id,
            "message": {
                "role": "user",
                "parts": [{"type": "text", "text": message}]
            }
        }
    }

    async with httpx.AsyncClient() as client:
        resp = await client.post(agent_url, json=payload)
        task = resp.json()["result"]

        # input-required: 추가 입력 요청
        while task["status"]["state"] == "input-required":
            agent_msg = task["status"]["message"]["parts"][0]["text"]
            print(f"Agent: {agent_msg}")
            user_input = input("You: ")

            # 같은 taskId + sessionId로 추가 메시지 전송
            followup = {
                "jsonrpc": "2.0", "id": 2,
                "method": "tasks/send",
                "params": {
                    "id": task_id,        # 동일한 task ID
                    "sessionId": session_id,
                    "message": {
                        "role": "user",
                        "parts": [{"type": "text", "text": user_input}]
                    }
                }
            }
            resp = await client.post(agent_url, json=followup)
            task = resp.json()["result"]

        return task
↑ 목차