A2A 에이전트 호출하기

LiteLLM을 통해 A2A 에이전트를 다양한 방법으로 호출하는 법을 알아봐요.

직접 에이전트로 테스트하고 싶다면? Google Gemini 기반의 템플릿 A2A 에이전트를 배포해 보세요: shin-bot-litellm/a2a-gemini-agent — 스트리밍을 지원하는 간단하고 배포 가능한 A2A 에이전트예요.

A2A SDK

A2A Python SDK(>= 1.1.0)를 사용해 A2A 프로토콜로 LiteLLM을 통해 에이전트를 호출할 수 있어요.

출처: 문서

pip install "a2a-sdk>=1.1.0,<2.0" httpx

에이전트에 protocolVersion: "1.0"을 고정하세요(권장). 그래야 응답이 1.x SDK와 일치합니다. 레거시 0.3 wire format을 원하면 "0.3"으로 고정하세요. Protocol versioning 문서를 참고하세요.

a2a-sdk 1.x는 A2AClient + dict MessageSendParams를 ClientFactory, protobuf Message / Part 타입, 그리고 stream 이벤트의 async generator인 send_message로 대체합니다. 아래 예시를 참고하세요.

비스트리밍 (Non-Streaming)

이 예시는 다음을 보여줘요:

사용 가능한 에이전트 나열 - /v1/agents를 조회해 키가 접근할 수 있는 에이전트 확인
에이전트 선택 - 목록에서 에이전트 선택
A2A로 호출 - A2A 프로토콜로 에이전트에 메시지 전송

import asyncio
from uuid import uuid4

import httpx
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from a2a.types import Message, Part, Role, SendMessageRequest
from a2a.utils.constants import TransportProtocol

# === CONFIGURE THESE ===
LITELLM_BASE_URL = "http://localhost:4000"  # Your LiteLLM proxy URL
LITELLM_VIRTUAL_KEY = "sk-<your-litellm-api-key>"             # Your LiteLLM Virtual Key
# =======================

def extract_text(parts) -> str:
    return "".join(getattr(p, "text", "") or "" for p in (parts or []))

def handle_event(event) -> None:
    populated = event.ListFields()
    if not populated:
        return
    field, value = populated[0]
    if field.name in ("message", "msg"):
        print(f"[message] {extract_text(value.parts)}")
    elif field.name == "task":
        print(f"[task {value.id}] {value.status.state}")

async def main():
    headers = {"Authorization": f"Bearer {LITELLM_VIRTUAL_KEY}"}

    async with httpx.AsyncClient(headers=headers, timeout=60.0) as http_client:
        # Step 1: List available agents
        response = await http_client.get(f"{LITELLM_BASE_URL}/v1/agents")
        agents = response.json()

        print("Available agents:")
        for agent in agents:
            print(f"  - {agent['agent_name']} (ID: {agent['agent_id']})")

        if not agents:
            print("No agents available for this key")
            return

        # Step 2: Select an agent and invoke it
        selected_agent = agents[0]
        agent_id = selected_agent["agent_id"]
        print(f"\nInvoking: {selected_agent['agent_name']}")

        # Step 3: Discover agent card and create a2a-sdk 1.x client
        base_url = f"{LITELLM_BASE_URL}/a2a/{agent_id}"
        resolver = A2ACardResolver(httpx_client=http_client, base_url=base_url)
        agent_card = await resolver.get_agent_card()

        config = ClientConfig(
            httpx_client=http_client,
            streaming=False,
            supported_protocol_bindings=[
                TransportProtocol.JSONRPC,
                TransportProtocol.HTTP_JSON,
            ],
        )
        client = ClientFactory(config).create(agent_card)

        msg = Message(
            message_id=uuid4().hex,
            role=Role.ROLE_USER,
            parts=[Part(text="Hello, what can you do?")],
        )
        request = SendMessageRequest(message=msg)

        async for event in client.send_message(request):
            handle_event(event)

if __name__ == "__main__":
    asyncio.run(main())

스트리밍 (Streaming)

a2a-sdk 1.x에서는 ClientConfig에 streaming=True를 설정하고 send_message를 반복하면 돼요. 같은 API가 스트리밍과 비스트리밍을 모두 처리합니다:

import asyncio
from uuid import uuid4

import httpx
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from a2a.types import Message, Part, Role, SendMessageRequest
from a2a.utils.constants import TransportProtocol

# === CONFIGURE THESE ===
LITELLM_BASE_URL = "http://localhost:4000"  # Your LiteLLM proxy URL
LITELLM_VIRTUAL_KEY = "sk-<your-litellm-api-key>"             # Your LiteLLM Virtual Key
LITELLM_AGENT_NAME = "ij-local"             # Agent name registered in LiteLLM
# =======================

async def main():
    base_url = f"{LITELLM_BASE_URL}/a2a/{LITELLM_AGENT_NAME}"
    headers = {"Authorization": f"Bearer {LITELLM_VIRTUAL_KEY}"}

    async with httpx.AsyncClient(headers=headers, timeout=60.0) as http_client:
        resolver = A2ACardResolver(httpx_client=http_client, base_url=base_url)
        agent_card = await resolver.get_agent_card()

        config = ClientConfig(
            httpx_client=http_client,
            streaming=True,
            supported_protocol_bindings=[
                TransportProtocol.JSONRPC,
                TransportProtocol.HTTP_JSON,
            ],
        )
        client = ClientFactory(config).create(agent_card)

        msg = Message(
            message_id=uuid4().hex,
            role=Role.ROLE_USER,
            parts=[Part(text="Tell me a long story")],
        )
        request = SendMessageRequest(message=msg)

        async for event in client.send_message(request):
            populated = event.ListFields()
            if populated:
                field, value = populated[0]
                if field.name in ("message", "msg"):
                    text = "".join(getattr(p, "text", "") or "" for p in value.parts)
                    print(text, end="", flush=True)
        print()

if __name__ == "__main__":
    asyncio.run(main())

/chat/completions API (OpenAI SDK)

익숙한 OpenAI SDK에 a2a/ 모델 접두사를 사용해 A2A 에이전트를 호출할 수도 있어요.

비스트리밍

Python
TypeScript
cURL

import openai

client = openai.OpenAI(
    api_key="sk-<your-litellm-api-key>",  # Your LiteLLM Virtual Key
    base_url="http://localhost:4000"  # Your LiteLLM proxy URL
)

response = client.chat.completions.create(
    model="a2a/my-agent",  # Use a2a/ prefix with your agent name
    messages=[
        {"role": "user", "content": "Hello, what can you do?"}
    ]
)

print(response.choices[0].message.content)

import OpenAI from 'openai';

const client = new OpenAI({
  apiKey: 'sk-<y...y>',  // Your LiteLLM Virtual Key
  baseURL: 'http://localhost:4000'  // Your LiteLLM proxy URL
});

const response = await client.chat.completions.create({
  model: 'a2a/my-agent',  // Use a2a/ prefix with your agent name
  messages: [
    { role: 'user', content: 'Hello, what can you do?' }
  ]
});

console.log(response.choices[0].message.content);

curl -X POST http://localhost:4000/v1/chat/completions \
  -H "Authorization: Bearer ***" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "a2a/my-agent",
    "messages": [
      {"role": "user", "content": "Hello, what can you do?"}
    ]
  }'

스트리밍

Python
TypeScript
cURL

import openai

client = openai.OpenAI(
    api_key="sk-<your-litellm-api-key>",  # Your LiteLLM Virtual Key
    base_url="http://localhost:4000"  # Your LiteLLM proxy URL
)

stream = client.chat.completions.create(
    model="a2a/my-agent",  # Use a2a/ prefix with your agent name
    messages=[
        {"role": "user", "content": "Tell me a long story"}
    ],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

import OpenAI from 'openai';

const client = new OpenAI({
  apiKey: 'sk-<y...y>',  // Your LiteLLM Virtual Key
  baseURL: 'http://localhost:4000'  // Your LiteLLM proxy URL
});

const stream = await client.chat.completions.create({
  model: 'a2a/my-agent',  // Use a2a/ prefix with your agent name
  messages: [
    { role: 'user', content: 'Tell me a long story' }
  ],
  stream: true
});

for await (const chunk of stream) {
  const content = chunk.choices[0]?.delta?.content;
  if (content) {
    process.stdout.write(content);
  }
}

curl -X POST http://localhost:4000/v1/chat/completions \
  -H "Authorization: Bearer ***" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "a2a/my-agent",
    "messages": [
      {"role": "user", "content": "Tell me a long story"}
    ],
    "stream": true
  }'

Task API ( tasks/get , tasks/list , …)

message/send가 submitted task를 반환하는 에이전트는 클라이언트가 tasks/get으로 폴링하기를 기대해요. 같은 LiteLLM base URL을 JSON-RPC로 호출하면 됩니다:

curl -X POST "http://localhost:4000/a2a/${AGENT_ID}" \
  -H "Authorization: Bearer ***" \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": "req-2",
    "method": "tasks/get",
    "params": {"id": "TASK_ID_FROM_SEND_RESPONSE"}
  }'

LiteLLM은 tasks/get, tasks/list, tasks/cancel, push-notification 메서드, 그리고 agent/getAuthenticatedExtendedCard를 업스트림 에이전트 URL로 전달합니다. 전체 메서드 목록은 Supported A2A methods 문서를 참고하세요.

주요 차이점

메서드	사용 사례	장점
A2A SDK	네이티브 A2A 프로토콜 통합	• 전체 A2A 프로토콜 지원 • task 상태 및 아티팩트 접근 • 컨텍스트 관리
OpenAI SDK	익숙한 OpenAI 스타일 인터페이스	• OpenAI 호출의 드롭인 대체 • LLM에서 에이전트 워크플로로의 쉬운 마이그레이션 • 기존 OpenAI 도구와 호환

OpenAI SDK를 사용할 때는 항상 에이전트 이름에 a2a/ 접두사를 붙이세요(예: a2a/my-agent). 그래야 요청이 LLM 프로바이더 대신 A2A 에이전트로 라우팅됩니다.

A2A 에이전트 호출하기