A2A 에러 처리

llamon SDK 는 에이전트 코드에서 raise 한 Python 예외를 A2A 프로토콜(0.3.x, a2a-sdk==0.3.26) 에러 포맷으로 자동 변환합니다. 평소처럼 raise 하면 응답 형식·이벤트 직렬화는 SDK 가 처리합니다.

핵심 4가지

기본은 그냥 raise — 매핑 테이블 이 Python 예외 → errorCode 변환
도메인 errorReason 필요하면 raise_application_error() — 매핑 테이블 우회
플로우 중간 노드 실패 — 이전 성공 노드의 artifacts 자동 보존, 추가 코드 불필요
스트리밍 중간 실패 — SDK 가 failed 자동 마감, 이중 artifact 발행 방지

두 레이어 — 언제 Protocol vs Application

레이어	언제 발생	클라이언트 수신 형태	개발자가 통제
Protocol-level	`new_task()` 전 거부	`error` 루트 필드 (JSON-RPC error)	제한적 — `_validate_request` 재정의, 혹은 SDK 가 판단
Application-level	에이전트 내부 처리 중 실패	`result.status.state="failed"` + `parts[data]`	개발자 주 제어 영역 — raise 한 Python 예외가 이 경로로 흐름

에이전트 코드에서 raise 한 예외는 거의 항상 Application-level 로 갑니다. Protocol-level 은 SDK 가 입력 검증·cancel·지원 경로 없음 등의 케이스에서 스스로 판단해 발행합니다.

errorCode — 분기용 고정 enum 8개

llamon_agent.core.errors.ErrorCode:

ErrorCode	의미
`INVALID_INPUT`	입력 누락·형식·스키마 불일치
`AUTH_REQUIRED`	인증·권한 필요 / 토큰 만료
`NOT_FOUND`	Task 만료·리소스 부재
`UPSTREAM_UNAVAILABLE`	MCP/RAG/DB/LLM 장애
`UPSTREAM_TIMEOUT`	외부 의존 타임아웃
`RATE_LIMITED`	쿼터·레이트 제한 (backoff 필요)
`GUARDRAIL_BLOCKED`	안전성·컴플라이언스 차단. `state:"failed"` + `artifacts[]` 폐기, 위반 메타는 `data.guardrail` sub-object (사용자 표시는 `violationCategory` 활용)
`INTERNAL_ERROR`	미상 오류 (fallback)

자동 매핑 테이블

에이전트 코드가 raise 하는 Python 예외 → errorCode 매핑 (python_exc_to_error_code()):

Python 예외	→ `errorCode`	`retriable` 기본
`GuardrailBlockedError`	`GUARDRAIL_BLOCKED`	false
`MissingAPIKeyError` · `RegistryUnauthorizedError`	`AUTH_REQUIRED`	false
`ConfigValidationError` · `MissingFieldError`	`INVALID_INPUT`	false
`RegistryNotFoundError`	`NOT_FOUND`	false
`RegistryTimeoutError` · `TimeoutError` · `asyncio.TimeoutError` · `httpx.TimeoutException`	`UPSTREAM_TIMEOUT`	true
`ComponentInvokeError` · `ComponentLoadError` · `httpx.ConnectError`	`UPSTREAM_UNAVAILABLE`	true
`httpx.HTTPStatusError` 429	`RATE_LIMITED`	true
`httpx.HTTPStatusError` 401·403	`AUTH_REQUIRED`	false
`langgraph.errors.GraphRecursionError`	`UPSTREAM_UNAVAILABLE` (`max_tool_iterations_exceeded`)	true
`MaxIterationError` · `MergeStrategyError` · `RoutingError` · `A2AProtocolError`	`INTERNAL_ERROR`	false
그 외 모든 예외	`INTERNAL_ERROR`	false

이 테이블에 없는 예외는 INTERNAL_ERROR 로 fallback. str(exc) 가 errorMessage 에, 타입 이름 기반 snake_case 가 errorReason 에 들어갑니다.

from llamon_agent.exceptions import ComponentInvokeError

async def check_bank_account(state):
    try:
        result = await mcp_client.call(...)
    except asyncio.TimeoutError:
        # → UPSTREAM_TIMEOUT + retriable:true 로 자동 변환됨
        raise  # 그대로 올리면 됨
    except httpx.HTTPStatusError as e:
        # → 429/401/403/5xx 에 따라 자동 매핑
        raise ComponentInvokeError(f"bank lookup failed: {e}") from e
    return result

MCP tool 자동 보호 (SDK 기본 동작)

MCPToolLoader().to_tools() 로 로드된 MCP 도구는 wrap_tool_output_normalizer 가 자동으로 적용되어, ReAct 루프에서 발생하는 두 종류의 실패를 사용자 코드 변경 없이 처리합니다:

실패 종류	동작	결과
LLM 인자 누락·타입 오류 (`pydantic.ValidationError`)	`Missing required argument: <field>` 한 줄로 변환 후 LLM 에 ToolMessage 로 반환	LLM 자가복구 (재시도) — Task 는 정상 진행
연결 실패 (`httpx.ConnectError`·`OSError ECONNREFUSED` 등) · 타임아웃 (`asyncio.TimeoutError`·`httpx.TimeoutException`)	같은 도구가 한 요청 내 누적 3회 실패 시 → `ComponentInvokeError` / `asyncio.TimeoutError` 로 escape	위 매핑 테이블에 따라 `UPSTREAM_UNAVAILABLE` / `UPSTREAM_TIMEOUT` (`retriable:true`) 로 변환

임계 미만에서는 LLM 이 다른 도구로 돌아가거나 재시도할 수 있게 sanitized text (Upstream tool '<name>' temporarily unavailable. Retry may succeed.) 가 반환됩니다. 매 요청 시작 시 카운터가 reset 되므로 다음 요청에 누적되지 않습니다.

ReAct 루프가 도구 호출 한도를 초과하면 LangGraph 가 GraphRecursionError 를 raise — 위 매핑 테이블에 의해 UPSTREAM_UNAVAILABLE (errorReason: max_tool_iterations_exceeded, retriable:true) 로 자동 변환됩니다.

자발적 에러 — `raise_application_error`

도메인 특화 errorReason 이나 retryAfterMs 같은 extras 를 지정하고 싶을 때 사용합니다. 매핑 테이블을 우회하고 exc 필드가 그대로 응답에 실립니다.

from llamon_agent.core.errors import ErrorCode, raise_application_error

async def classify_query(state):
    query = state.get("query", "")
    if not query:
        raise_application_error(
            ErrorCode.INVALID_INPUT,
            "query_empty",
            message="질의 본문이 비어있습니다.",
            retriable=False,
        )

    if await rate_limiter.exceeded():
        raise_application_error(
            ErrorCode.RATE_LIMITED,
            "classifier_quota_exceeded",
            message="분류 서비스 일일 할당량 초과",
            retriable=True,
            retryAfterMs=60_000,
        )

extras (keyword-only arguments 뒤의 **kwargs) 는 ApplicationErrorData 의 optional 필드로 통과됩니다. 허용되는 키: retryAfterMs·errorSource·domain·node·flow 등.

플로우 중간 노드 실패 — 자동 artifacts 보존

플로우 에이전트에서 중간 노드가 실패하면 SDK 가 LangGraph checkpoint 를 사후 조회해:

실패 노드 id (state.next) → data.node.id
GraphBuilder 에서 지정한 node_kind → data.node.kind
완료된 노드 목록 (history.metadata.writes) → data.flow.completedNodes
완료 노드들의 산출물 → failed 이전에 artifact-update 이벤트로 선행 발행

개발자는 추가 작업 없이 GraphBuilder.node(..., node_kind="...") 만 지정하면 됩니다.

from llamon_agent import GraphBuilder, END, START

builder = (
    GraphBuilder()
    .node("fetch_subject", fetch_subject_node, node_kind="business")
    .node("rag_search", rag_search_node, node_kind="business")
    .node("generate_response", generate_response_node, node_kind="llm")
    .edge(START, "fetch_subject")
    .edge("fetch_subject", "rag_search")
    .edge("rag_search", "generate_response")
    .edge("generate_response", END)
)

rag_search 가 ComponentInvokeError 를 raise 하면:

클라이언트는 data.errorCode="UPSTREAM_UNAVAILABLE" + data.node={"id":"rag_search","kind":"business"} + data.flow.completedNodes=["fetch_subject"] 수신
artifacts[] 에 fetch_subject 의 결과가 보존됨 → 부분 결과 UI 가능

스트리밍 중간 실패 — 자동 마감

message/stream 에서 에이전트가 청크를 일부 발행한 뒤 예외를 던지면:

SDK 가 StreamPartialFailure 로 포획
invoke 폴백 금지 (이중 artifact 방지)
현재까지의 text/data/files 버퍼를 best-effort 로 flush
failed status-update 이벤트로 마감 (text + data Part 동봉)

from llamon_agent.exceptions import ComponentInvokeError

async def stream_response(state, *, agent_url):
    writer = get_stream_writer()
    full = ""
    async for chunk in call_agent_stream(agent_url, build_agent_context(state)):
        writer(chunk)
        full += chunk
        if some_mid_stream_error_condition:
            # 청크 N개 발행 후 실패 — SDK 가 자동으로 failed 마감
            raise ComponentInvokeError("mid-stream failure")
    return {"output": full}

스트림 진입 전(첫 청크 발행 전) 실패는 기존 invoke 폴백 경로로 처리됩니다. 에이전트가 stream 외 invoke 도 구현해 두면 자연스럽게 재시도됩니다.

값 규약 — `errorReason` 등 포맷

SDK 는 응답 직렬화 단계에서 Pydantic validator 로 errorSource·domain 같은 포맷 고정 필드를 엄격하게 검증합니다. 위반 시 fallback (INTERNAL_ERROR + error_data_build_failed reason) 으로 교체되어 응답은 나가지만, 개발자 의도는 사라지므로 규약을 지킵시다. errorReason 은 형식이 강제되지 않지만(최소 non-empty + ASCII) 팀 일관성을 위해 권장 포맷을 따르는 것을 강력 권고합니다.

항목	규약
`errorReason`	SDK 강제: non-empty + ASCII. 권장: `snake_case_verb_noun` (`account_owner_check_failed`). snake_case 는 validator 가 강제하지 않으며 팀 컨벤션·PR 리뷰에서 관리. 분기 용도 아님 — 로그·대시보드 그룹핑·i18n 키 용도
`errorSource`	`"{system}:{locator}"` 고정 포맷. system ∈ `mcp` · `rag` · `llm` · `db` · `http` · `validator` · `guardrail` · `internal`. 예: `"mcp:MCP-001/check_account_owner"`
`domain`	`kebab-case`, ASCII 만. 에이전트 식별자와 일치 (`subject-management`)
`retried`	서버 내부 재시도 횟수, 0 부터
`retryAfterMs`	정수 ms. `RATE_LIMITED` 권장, 없으면 클라가 exponential backoff

외부 에러 코드 체계와의 관계

설계서·사내 스펙에 번호 코드 체계(DOC_001·E1001 등)가 이미 있다면 A2A 에러 스키마와의 관계를 먼저 정리해야 합니다. 결론부터: 대부분의 경우 클라이언트로 나가는 최종 응답에 번호 코드를 별도 슬롯으로 실을 필요는 없습니다 — 아래 SDK 필드로 흡수됩니다.

용도	SDK 필드	포맷
클라이언트 주요 분기 (재시도·입력폼·로그인)	`errorCode` (8종 enum)	고정 어휘
집계·i18n 키·대시보드 그룹핑	`errorReason`	권장 snake_case (ASCII 필수, 그 외 강제 없음)
도메인 코드 그룹핑 (trace 최상위 promote)	`domainCode`	카탈로그 키 (`DOC_005`)
에이전트 식별 (중의성 해소)	`domain`	kebab-case
사용자 표시 문구	`errorMessage`	자연어

외부 체계의 “카테고리”는 errorCode 8종에, “구체 시나리오”는 errorReason(또는 전용 domainCode 필드)에 매핑합니다. 같은 번호가 에이전트마다 다른 의미를 가지면 domain 으로 식별합니다.

Task 상태 의미론 — “에러 아닌 결과”

외부 카탈로그에 “적합성 미충족”·“관련 정보 없음” 같이 정상 완료이되 특수 안내 유형이 있다면 state:"failed" 가 아닌 다음 상태로 표현하세요. failed 로 보내면 Task 가 종료돼 HITL 재개·후속 턴을 이어 붙일 수 없습니다.

input-required — 사용자 추가 입력 필요
completed + data Part — 정상 완료이되 부적합 사유·빈 결과 안내

HITL 처리 상세는 HITL 과 분리 섹션 참조.

체크리스트

도메인 번호 코드는 카탈로그 + 팩토리로 일원화 — errorCode · errorReason · domainCode 매핑을 한곳에
domain 에 에이전트 식별자(kebab-case) 지정
“에러 아닌 결과”(빈 결과·부적합 안내)는 failed 대신 input-required / completed 로
가드레일 차단은 carve-out — failed + GUARDRAIL_BLOCKED 유지 (위 caution 박스 참조)

HITL 과 분리

input-required · auth-required 는 에러 아닙니다 — 정상 플로우 제어입니다. SDK 의 invoke_with_hitl() / _emit_input_required() 경로가 담당하며 Phase 3 의 try/except 에서 포획되지 않습니다.

from llamon_agent.core.hitl import HITLInterrupt

async def ask_birthday(state):
    birthday = state.get("birthday")
    if not birthday:
        # raise 하지 않고 return — SDK 가 input-required 로 전송
        return HITLInterrupt(
            payload={"question": "생년월일을 알려주세요."},
            thread_id=state["thread_id"],
        )
    ...

SDK 내부 동작 요약

에이전트 개발자가 알아두면 좋은 내부 흐름:

agent code raises Python exc
        ↓
CustomAgentExecutor.execute()
        │
        ├─ ServerError        → Protocol-level (data 이미 주입됨), 그대로 전파
        ├─ Exception          → _enqueue_failed_response(exc)
        │                        │
        │                        ├─ isinstance(exc, LlamonApplicationError)?
        │                        │    Yes → exc.code / exc.reason / exc.extra 그대로 사용
        │                        │    No  → python_exc_to_error_code(exc) 매핑 테이블
        │                        │
        │                        ├─ extract_flow_error_context(agent, thread_id)
        │                        │    (best-effort checkpoint 조회)
        │                        │
        │                        ├─ intermediate artifacts → updater.add_artifact()
        │                        └─ updater.failed(message=text+data)
        │
        └─ StreamPartialFailure → _enqueue_failed_response (폴백 금지)

테스트에서 에러 응답 검증

Application-level 에러 응답을 유닛 테스트에서 검증할 때 패턴:

import pytest
from a2a.types import DataPart, TaskStatusUpdateEvent
from llamon_agent.exceptions import ComponentInvokeError

from .conftest import stub_executor_run  # 프로젝트 fixture


@pytest.mark.asyncio
async def test_mcp_failure_returns_upstream_unavailable():
    events = await stub_executor_run(
        agent_dispatch_raises=ComponentInvokeError("MCP down"),
    )

    failed = next(
        e for e in reversed(events)
        if isinstance(e, TaskStatusUpdateEvent) and e.status.state.value == "failed"
    )
    data = next(
        p.root.data for p in failed.status.message.parts if isinstance(p.root, DataPart)
    )

    assert data["errorCode"] == "UPSTREAM_UNAVAILABLE"
    assert data["retriable"] is True

SDK 내부 테스트 예시: tests/unit/test_executor_errors_application.py.

공개 API 요약

from llamon_agent.core.errors import (
    # 열거형
    ErrorCode,                     # 8종 enum
    JSON_RPC_CODE_TO_ERROR_CODE,   # JSON-RPC 숫자 → errorCode 매핑

    # Pydantic 모델
    BaseErrorData,                 # 공통 베이스
    ProtocolErrorData,             # error.data 위치
    ApplicationErrorData,          # result.status.message.parts[data].data 위치
    ErrorNode, ErrorFlow,          # 중첩 서브모델

    # 매핑
    python_exc_to_error_code,      # (exc) -> (code, reason, retriable)
    json_rpc_to_error_code,        # (jsonrpc_code: int) -> ErrorCode

    # 팩토리
    make_protocol_error_data,
    make_application_error_data,
    application_error_to_parts,        # (data, display_text) -> [TextPart, DataPart]
    build_guardrail_application_error, # 가드레일 차단 → ApplicationErrorData (SDK 내부에서 사용)
    build_guardrail_extras,            # 가드레일 extras (errorSource + guardrail sub-object)

    # 개발자 helper
    LlamonApplicationError,        # 커스텀 예외 클래스
    raise_application_error,       # 한 줄 helper
    make_domain_error_raiser,      # 카탈로그 → 전용 raise 헬퍼 팩토리 (도메인 코드 매핑)
    DomainErrorSpec,               # 카탈로그 항목 TypedDict ({sdk_code, reason, title, retriable})
)

대부분의 에이전트 코드는 ErrorCode · raise_application_error 만 씁니다. 도메인 번호 코드를 카탈로그로 묶어 쓴다면 make_domain_error_raiser + DomainErrorSpec — 패턴은 도메인 에러 매핑 가이드 참조. 나머지는 SDK 내부 구현용으로, 가드레일 팩토리(build_guardrail_application_error / build_guardrail_extras)는 커스텀 가드레일 백엔드를 직접 구현할 때만 사용합니다.