[A2A 알아보기 2편] A2A Python SDK: 나의 에이전트를 다른 사람의 에이전트가 호출할 수 있도록 하기 위해 A2A 서버로 공개하기

A2A Python SDK 소개

A2A Python SDK는 파이썬으로 만든 에이전트를 A2A(Agent2Agent) 프로토콜 (:pytorch::kr: [A2A 알아보기 1편] A2A 프로토콜: 프레임워크가 서로 달라도 에이전트들끼리 서로 작업을 위임하게 하는 표준 규격)을 따르는 서버로 공개하게 해 주는 공식 라이브러리입니다. 이미 동작하는 에이전트를 가진 팀이 그것을 다른 팀의 에이전트에게 열어 주려고 하면, 할 일이 에이전트 로직 바깥에 몰려 있습니다. 능력을 기술한 문서를 정해진 위치에 올려야 하고, 요청을 받아 태스크로 만들어 상태를 관리해야 하고, 진행 상황을 스트리밍으로 흘려 보내야 합니다. 이 부분을 직접 구현하면 에이전트 자체보다 주변 코드가 더 커집니다.

A2A Python SDK는 그 주변부를 맡습니다. 개발자가 채우는 것은 두 가지로, 에이전트가 무엇을 할 수 있는지 적은 AgentCard와 실제 일을 수행하는 AgentExecutor 구현입니다. 나머지는 SDK가 제공하는 요청 핸들러와 라우트 생성 함수가 처리합니다. 라우트는 StarletteFastAPI 같은 ASGI(Asynchronous Server Gateway Interface) 프레임워크에 그대로 붙일 수 있어서, 인증과 로깅 같은 기존 미들웨어 구성을 유지한 채 A2A 엔드포인트만 추가할 수 있습니다.

본 게시물은 a2a-samples 저장소helloworld 예제를 기준으로 SDK의 구성 요소를 하나씩 짚고, 서버를 띄운 뒤 실제로 오간 JSON까지 확인합니다. 아래에 실린 출력은 모두 a2a-sdk 1.1.0과 Python 3.11에서 직접 실행해 받은 것입니다.

A2A Python SDK가 대신 처리하는 것

A2A 서버를 직접 구현한다고 하면 프로토콜 명세에서 옮겨 와야 하는 항목이 적지 않습니다. 에이전트 카드를 잘 알려진 경로로 제공하는 일, 들어온 메시지를 태스크로 바꾸고 식별자를 발급하는 일, 태스크의 상태 전이를 관리하는 일, 스트리밍 이벤트를 순서대로 내보내는 일, 선언하지 않은 기능을 호출받았을 때 정해진 오류를 돌려주는 일이 모두 포함됩니다. SDK는 이 항목들을 계층으로 나눠 맡아, 개발자가 맨 아래 한 칸만 채우면 되도록 합니다:

지원 범위는 저장소가 표로 명시하고 있으며, 1.0 명세를 세 가지 바인딩 모두에서 지원하고 0.3에 대해서는 호환 모드를 제공합니다:

명세 버전 전송 방식 클라이언트 서버
1.0 JSON-RPC 지원 지원
1.0 HTTP+JSON/REST 지원 지원
1.0 gRPC 지원 지원
0.3(호환 모드) JSON-RPC 지원 지원
0.3(호환 모드) HTTP+JSON/REST 지원 지원
0.3(호환 모드) gRPC 지원 지원

선택 의존성으로 gRPC, OpenTelemetry 기반 추적, 그리고 PostgreSQL과 MySQL, SQLite를 쓰는 SQL 태스크 저장소를 함께 제공합니다. 예제에서 쓰는 메모리 기반 저장소는 프로세스가 내려가면 태스크가 사라지므로, 오래 사는 태스크를 다루려면 이 SQL 저장소로 바꿔야 합니다.

A2A Python SDK 실습 준비

Python 3.10 이상이 필요합니다. 예제 저장소를 받고 가상 환경에 SDK를 설치합니다:

git clone https://github.com/a2aproject/a2a-samples.git -b main --depth 1
cd a2a-samples

python -m venv .venv
source .venv/bin/activate          # Windows는 .venv\Scripts\activate

pip install -r samples/python/agents/helloworld/requirements.txt

공식 튜토리얼은 저장소 최상위의 samples/python/requirements.txt를 설치하라고 안내하는데, 그 파일은 a2a-sdk[http-server]>=0.3.0처럼 하한만 지정합니다. 반면 helloworld 예제의 requirements.txta2a-sdk==1.1.0으로 버전을 고정하고 sse-starlette까지 포함하므로, 예제를 그대로 재현하려면 이쪽이 안전합니다. 설치가 끝났는지는 임포트 한 줄로 확인합니다:

python -c "import a2a; print('A2A SDK imported successfully')"

A2A Python SDK의 에이전트 카드 정의

에이전트가 할 수 있는 일은 AgentSkill 객체로 하나씩 기술합니다. helloworld 예제는 받은 메시지를 그대로 되돌려 주는 기능 하나를 선언합니다:

skill = AgentSkill(
    id='echo_bot',
    name='Echo Bot',
    description='An example agent that acknowledges client request and responds with a "Hello World" message.',
    input_modes=['text/plain'],
    output_modes=['text/plain'],
    tags=['a2a', 'echo-example'],
    examples=['hi', 'how are you'],
)

공식 튜토리얼 문서의 설명 문장은 이 기능을 "Returns hello world" 라는 이름의 hello_world 기능이라고 적고 있지만, 저장소의 실제 코드는 위와 같이 echo_botEcho Bot을 씁니다. 문서 산문이 코드 변경을 따라오지 못한 것으로, 예제를 따라 하다 이름이 맞지 않아 혼란스러울 때는 저장소의 __main__.py를 기준으로 삼으면 됩니다.

기능을 모아 에이전트 카드를 만듭니다. 카드에는 정체성 정보와 기본 입출력 미디어 타입, 지원 기능, 그리고 접속 가능한 인터페이스 목록이 들어갑니다:

public_agent_card = AgentCard(
    name='Hello World Agent',
    description='Just a hello world agent',
    version='0.0.1',
    default_input_modes=['text/plain'],
    default_output_modes=['text/plain'],
    capabilities=AgentCapabilities(streaming=True, extended_agent_card=True),
    supported_interfaces=[
        AgentInterface(
            protocol_binding='JSONRPC',
            url='http://127.0.0.1:9999',
            protocol_version='1.0',
        )
    ],
    skills=[skill],
)

supported_interfacesprotocol_version='1.0'이 이 에이전트가 말하는 프로토콜 버전이고, protocol_binding에는 JSONRPC, GRPC, HTTP+JSON 중 하나가 들어갑니다. 파이썬 쪽 필드 이름이 snake_case인 것과 달리 실제로 공개되는 JSON은 camelCase이므로, 코드의 default_input_modes는 카드에서 defaultInputModes로 나타납니다.

예제는 인증한 클라이언트에게만 보여 줄 확장 카드도 함께 정의합니다. 공개 카드에는 echo_bot 하나만 넣고, 확장 카드에는 echo_bot_super_mode라는 기능을 더해 두 개를 실어 보냅니다. 어떤 기능은 아무나 보게 하고 어떤 기능은 계약을 맺은 상대에게만 알리는 상황을 이 구조로 표현할 수 있습니다.

A2A Python SDK의 실행기 구현

실제 작업은 AgentExecutor 인터페이스를 구현한 클래스가 수행합니다. execute 메서드 하나가 요청 하나를 처리하며, 태스크를 만들고 상태를 갱신하고 결과물을 붙이는 흐름이 그 안에 모두 들어갑니다:

class HelloWorldAgentExecutor(AgentExecutor):
    def __init__(self) -> None:
        self.agent = HelloWorldAgent()

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        # 1. 요청 맥락에서 태스크를 가져오고, 없으면 새로 만들어 큐에 넣습니다
        if context.current_task:
            task = context.current_task
        else:
            task = new_task_from_user_message(context.message)
            await event_queue.enqueue_event(task)

        # 2. TaskUpdater로 상태를 '작업 중'으로 바꿉니다
        task_updater = TaskUpdater(
            event_queue=event_queue, task_id=task.id, context_id=task.context_id
        )
        await task_updater.update_status(
            state=TaskState.TASK_STATE_WORKING,
            message=new_text_message('Processing request...'),
        )

        # 3. 사용자 입력을 꺼내 에이전트를 호출합니다
        query = get_message_text(context.message)
        result = await self.agent.invoke(user_request=query) if query else 'No text input is provided!'

        # 4. 생성한 응답을 아티팩트로 붙입니다
        await task_updater.add_artifact(parts=[new_text_part(text=result, media_type='text/plain')])

        # 5. 상태를 '완료'로 바꿉니다
        await task_updater.update_status(
            state=TaskState.TASK_STATE_COMPLETED,
            message=new_text_message('Request is completed!'),
        )

여기서 주목할 부분은 결과를 메시지가 아니라 아티팩트로 붙인다는 점입니다. A2A 명세는 메시지를 대화의 한 턴으로, 아티팩트를 태스크의 산출물로 구분하고 결과를 아티팩트로 돌려주라고 권고합니다. 메시지는 태스크 기록에 보존된다는 보장이 없기 때문입니다. TaskUpdater는 이 규칙에 맞춰 상태 갱신 이벤트와 아티팩트 갱신 이벤트를 각각 만들어 큐에 넣어 줍니다.

cancel 메서드도 인터페이스에 포함되어 있습니다. helloworld 예제는 취소를 지원하지 않아 NotImplementedError를 던지는데, 실제 에이전트를 만들 때는 진행 중인 작업을 어떻게 중단할지 여기에 구현해야 합니다.

A2A Python SDK의 서버 구성

요청 핸들러가 실행기와 태스크 저장소, 그리고 두 종류의 카드를 묶습니다. 그다음 라우트 생성 함수로 엔드포인트를 만들어 ASGI 앱에 등록합니다:

request_handler = DefaultRequestHandler(
    agent_executor=HelloWorldAgentExecutor(),
    task_store=InMemoryTaskStore(),
    agent_card=public_agent_card,
    extended_agent_card=extended_agent_card,
)

routes = []
routes.extend(create_agent_card_routes(public_agent_card))
routes.extend(create_jsonrpc_routes(request_handler, '/'))

app = Starlette(routes=routes)
uvicorn.run(app, host='127.0.0.1', port=9999)

create_agent_card_routes는 카드를 /.well-known/agent-card.json 경로로 공개하는 라우트를, create_jsonrpc_routes는 JSON-RPC 메서드 호출을 받아 핸들러로 넘기는 라우트를 돌려줍니다. JSON-RPC 대신 REST로 서비스하려면 create_rest_routes를 쓰면 되고, 두 가지를 같은 앱에 함께 붙일 수도 있습니다. 핸들러에 카드를 넘기는 이유는 선언된 기능을 검사하기 위해서입니다. 카드에 streaming=False인 에이전트가 스트리밍 요청을 받으면 핸들러가 정해진 오류를 돌려줍니다.

서버는 아래 명령으로 띄웁니다:

python samples/python/agents/helloworld/__main__.py

A2A Python SDK로 만든 서버의 실제 응답

서버가 뜨면 클라이언트가 가장 먼저 하는 일은 에이전트 카드를 받아 오는 것입니다:

curl로 그 경로를 직접 열어 보면 파이썬 코드에 적은 값들이 camelCase JSON으로 바뀌어 나옵니다:

curl -s http://127.0.0.1:9999/.well-known/agent-card.json
{
  "name": "Hello World Agent",
  "description": "Just a hello world agent",
  "supportedInterfaces": [
    { "url": "http://127.0.0.1:9999", "protocolBinding": "JSONRPC", "protocolVersion": "1.0" }
  ],
  "version": "0.0.1",
  "capabilities": { "streaming": true, "extendedAgentCard": true },
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["text/plain"],
  "skills": [
    {
      "id": "echo_bot",
      "name": "Echo Bot",
      "description": "An example agent that acknowledges client request and responds with a \"Hello World\" message.",
      "tags": ["a2a", "echo-example"],
      "examples": ["hi", "how are you"],
      "inputModes": ["text/plain"],
      "outputModes": ["text/plain"]
    }
  ]
}

공개 카드에 echo_bot_super_mode가 보이지 않는 것으로 확장 카드 구분이 실제로 동작하고 있음을 확인할 수 있습니다. 이제 메시지를 보내 봅니다. SDK 없이 JSON-RPC 요청을 그대로 만들어 보면 프로토콜의 구조가 한눈에 들어옵니다:

curl -s -X POST http://127.0.0.1:9999/ \
  -H "Content-Type: application/json" \
  -H "A2A-Version: 1.0" \
  -d '{
    "jsonrpc": "2.0",
    "id": "1",
    "method": "SendMessage",
    "params": {
      "message": {
        "messageId": "msg-1",
        "role": "ROLE_USER",
        "parts": [{"text": "Say hello."}]
      }
    }
  }'

돌아온 응답에는 서버가 발급한 태스크 식별자와 대화 식별자, 종료 상태, 아티팩트, 그리고 오간 메시지 기록이 함께 들어 있습니다:

{
  "jsonrpc": "2.0",
  "id": "1",
  "result": {
    "task": {
      "id": "263147b6-bdfd-4910-810b-eb0dd0b970ec",
      "contextId": "77f62894-9bd6-4728-b03a-1ace6158c2b6",
      "status": {
        "state": "TASK_STATE_COMPLETED",
        "message": { "role": "ROLE_AGENT", "parts": [{ "text": "Request is completed!" }] },
        "timestamp": "2026-09-11T03:07:54.235547Z"
      },
      "artifacts": [
        {
          "artifactId": "b30d1c8e-78a7-4aab-9d1e-ef6ac9e9490c",
          "parts": [
            { "text": "Hello, World! I have received your request (Say hello.)", "mediaType": "text/plain" }
          ]
        }
      ],
      "history": [
        { "messageId": "msg-1", "role": "ROLE_USER", "parts": [{ "text": "Say hello." }] },
        { "role": "ROLE_AGENT", "parts": [{ "text": "Processing request..." }] }
      ]
    }
  }
}

파트가 {"text": "..."} 형태이고 kind 판별자 필드가 없다는 점이 1.0 명세를 따른 결과입니다. 0.3에서는 같은 자리에 {"kind": "text", "text": "..."} 가 왔습니다. 열거형 값이 TASK_STATE_COMPLETEDROLE_USER처럼 대문자로 나오는 것도 명세가 정한 ProtoJSON 직렬화 규칙입니다.

같은 메시지를 스트리밍으로 보내면 응답이 Server-Sent Events 네 건으로 나뉘어 도착합니다. 메서드 이름만 SendStreamingMessage로 바꾸면 됩니다:

data: {"result": {"task": {"id": "9dddbeb7-...", "status": {"state": "TASK_STATE_SUBMITTED"}, ...}}, "id": "2", "jsonrpc": "2.0"}

data: {"result": {"statusUpdate": {"taskId": "9dddbeb7-...", "status": {"state": "TASK_STATE_WORKING", "message": {"role": "ROLE_AGENT", "parts": [{"text": "Processing request..."}]}}}}, "id": "2", "jsonrpc": "2.0"}

data: {"result": {"artifactUpdate": {"taskId": "9dddbeb7-...", "artifact": {"parts": [{"text": "Hello, World! I have received your request (stream please)", "mediaType": "text/plain"}]}}}, "id": "2", "jsonrpc": "2.0"}

data: {"result": {"statusUpdate": {"taskId": "9dddbeb7-...", "status": {"state": "TASK_STATE_COMPLETED", "message": {"role": "ROLE_AGENT", "parts": [{"text": "Request is completed!"}]}}}}, "id": "2", "jsonrpc": "2.0"}

첫 이벤트가 태스크 객체이고, 그다음이 상태 갱신과 아티팩트 갱신이며, 종료 상태에 도달한 상태 갱신을 끝으로 스트림이 닫힙니다. 실행기 코드에서 update_statusadd_artifact를 부른 순서가 그대로 네 건의 이벤트가 되었다는 점을 코드와 나란히 놓고 보면 이해하기 쉽습니다. 상태 갱신 이벤트가 statusUpdate로, 아티팩트 갱신 이벤트가 artifactUpdate로 감싸여 있는 것 역시 1.0에서 바뀐 표현 방식입니다.

A2A Python SDK의 클라이언트 코드

직접 JSON을 만들지 않고 SDK의 클라이언트를 쓰면 카드 조회와 전송 방식 선택을 대신 처리해 줍니다. A2ACardResolver가 잘 알려진 경로에서 카드를 받아 오고, create_client가 그 카드를 보고 클라이언트를 만듭니다:

async with httpx.AsyncClient() as httpx_client:
    resolver = A2ACardResolver(
        httpx_client=httpx_client,
        base_url='http://127.0.0.1:9999',
    )
    public_agent_card = await resolver.get_agent_card()

config = ClientConfig(streaming=False)
client = await create_client(agent=public_agent_card, client_config=config)

message = new_text_message('Hi there', role=Role.ROLE_USER)
request = SendMessageRequest(message=message)

async for chunk in client.send_message(request):
    print(chunk)

await client.close()

ClientConfig(streaming=True)로 바꾸면 같은 send_message 호출이 앞에서 본 네 건의 이벤트를 차례로 내놓습니다. 즉 스트리밍 여부가 별도 메서드가 아니라 설정값으로 갈립니다. 예제에 포함된 test_client.py는 서버를 직접 띄웠다 내리는 pytest 픽스처를 포함하고 있어서, 서버를 따로 실행하지 않고 pytest로 한 번에 확인할 수도 있습니다.

LLM을 붙인 더 복잡한 예제로는 같은 저장소의 langgraph 예제가 있습니다. LangGraph로 만든 환율 변환 에이전트가 질문이 모호할 때 태스크를 입력 대기 상태로 두고 되물으며, 클라이언트가 앞선 응답의 태스크 식별자와 대화 식별자를 붙여 다시 메시지를 보내 같은 태스크를 이어 갑니다. 여러 턴에 걸친 대화가 A2A에서 어떻게 표현되는지가 이 예제에 드러납니다.

다만 이 예제는 0.3 시절의 SDK API로 작성되어 있어 위에서 설치한 1.1.0 환경에서는 임포트 단계에서 멈춥니다. 예제의 pyproject.toml이 요구하는 버전이 a2a-sdk>=0.3.0으로 하한만 잡혀 있어 설치는 최신 버전으로 되는데, 예제 코드가 불러오는 a2a.server.apps 모듈은 1.1.0에 없고 AgentCard의 최상위 url 필드도 제거되었기 때문입니다. 상태 값 표기도 TaskState.input_required처럼 구 형식이라 1.0 명세의 TASK_STATE_INPUT_REQUIRED와는 다릅니다. 다중 턴 흐름의 설계를 참고하는 자료로는 여전히 쓸모가 있지만, 실행까지 하려면 예제가 갱신되기를 기다리거나 예제에 맞는 구버전 SDK를 별도 환경에 설치해야 합니다. 직접 만든 0.3 코드를 옮기는 경우라면 SDK 저장소의 v0.3에서 v1.0으로 넘어가는 마이그레이션 가이드가 타입 변경부터 요청 핸들러와 서버 구성 방식, 클라이언트 생성까지 항목별로 정리해 두었습니다.

A2A Python SDK는 누구에게 맞는가

이미 파이썬으로 동작하는 에이전트를 가지고 있고 그것을 다른 팀이나 외부에 열어 줄 계획이 있다면, A2A Python SDK는 프로토콜 구현을 직접 옮겨 적는 대신 쓸 수 있는 가장 짧은 경로입니다. 기존 에이전트 로직은 AgentExecutor.execute 안에서 호출하기만 하면 되므로 프레임워크를 바꿀 필요가 없고, 라우트를 기존 ASGI 앱에 붙이는 방식이라 인증과 로깅 구성도 그대로 유지됩니다.

반대로 에이전트를 외부에 공개할 계획이 아직 없고 도구 연결만 필요한 단계라면 이 SDK를 도입할 이유가 없습니다. 그 경우에는 MCP를 쓰는 편이 맞고, A2A 프로토콜 자신도 도구 호출은 자신의 영역이 아니라고 밝히고 있습니다.

운영을 염두에 둔다면 두 가지를 미리 확인해 두는 것이 좋습니다. 하나는 태스크 저장소로, 예제의 InMemoryTaskStore는 프로세스와 함께 사라지므로 오래 걸리는 태스크를 다루려면 SDK가 함께 제공하는 DatabaseTaskStore로 바꾸고 a2a-sdk[postgresql]처럼 해당 데이터베이스용 추가 의존성을 설치해야 합니다. 다른 하나는 입력 신뢰 문제로, A2A 프로젝트는 외부 에이전트가 보낸 에이전트 카드와 메시지, 아티팩트를 모두 신뢰할 수 없는 입력으로 다루라고 예제에 함께 적어 두었습니다. 카드의 설명 필드에 넣은 문구가 그대로 LLM 프롬프트로 들어가면 프롬프트 주입(Prompt Injection)의 통로가 되기 때문입니다.

A2A Python SDK의 라이선스

A2A Python SDK는 Apache License 2.0으로 공개되어 있어 개인 및 상업적 목적으로 자유롭게 사용할 수 있습니다. 본 게시물에서 사용한 예제가 담긴 a2a-samples 저장소도 같은 라이선스를 따릅니다.

:books: A2A Python 퀵스타트 튜토리얼 문서

:github: A2A Python SDK GitHub 저장소

:file_folder: A2A 예제 코드 저장소

더 읽어보기




이 글은 GPT 모델로 정리한 초안을 바탕으로 한 것으로, 원문의 내용 또는 의도와 다르게 정리된 내용이 있을 수 있습니다. 관심있는 내용이시라면 원문도 함께 참고해주세요! 읽으시면서 어색하거나 잘못된 내용을 발견하시면 댓글로 알려주시기를 부탁드립니다. :hugs:

:pytorch:파이토치 한국 사용자 모임:south_korea:은 이런 글들을 한국어로 정리해 나누고 있습니다. 회원으로 가입하시면 주요 글들을 이메일:love_letter:로 보내드리고, 텔레그램(Telegram)Slack/Discord/Teams/Dooray/GoogleChat 등으로도 새 글 알림을 받으실 수 있습니다. :smiley:

:wrapped_gift: 아래:down_right_arrow:쪽에 좋아요:+1:를 눌러주시면 다음 글을 정리하는 데 힘이 됩니다~ :star_struck: