6. 서버와 상호작용하기
6. 서버와 상호작용하기
Helloworld A2A 서버가 실행 중이니, 이제 여기에 요청을 보내볼게요.
출처: 문서
본문
Helloworld 테스트 클라이언트
test_client.py 스크립트는 다음을 보여줘요.
- 서버에서 Agent Card를 가져오기.
create_client를 사용해 클라이언트 만들기.Send Message와Send Streaming Message요청 모두 보내기.
새 터미널 창을 열고 가상 환경을 활성화한 뒤 a2a-samples 디렉터리로 이동해요.
가상 환경을 활성화해요(만든 디렉터리에서 해야 해요):
Mac/Linux:
source .venv/bin/activate
Windows:
.venv\Scripts\activate
테스트 클라이언트를 실행해요:
# from the a2a-samples directory
python samples/python/agents/helloworld/test_client.py
클라이언트 코드 이해하기
test_client.py의 핵심 부분을 살펴볼게요.
Agent Card 가져오기:
import httpx # noqa: PLC0415
from a2a.client import A2ACardResolver # noqa: PLC0415
# Initializes the A2ACardResolver instance with an HTTP client, base URL,
# and uses the default path for the agent card.
async with httpx.AsyncClient() as httpx_client:
resolver = A2ACardResolver(
httpx_client=httpx_client,
base_url='http://127.0.0.1:9999',
# Provide agent_card_path, if your agent uses a different path
# agent_card_path='' # noqa: ERA001
)
public_agent_card = await resolver.get_agent_card()
A2ACardResolver 클래스는 편의 기능이에요. get_agent_card()가 호출되면 제공된 base URL을 기반으로 서버의 /.well-known/agent-card.json 엔드포인트에서 AgentCard를 가져오고, 이를 사용해 클라이언트를 초기화해요.
클라이언트 초기화 & 비스트리밍 메시지 보내기:
from a2a.client import ClientConfig, create_client # noqa: PLC0415
from a2a.helpers import new_text_message # noqa: PLC0415
from a2a.types import Role, SendMessageRequest # noqa: PLC0415
print('\nInitializing a non-streaming client.')
config = ClientConfig(streaming=False)
client = await create_client(agent=public_agent_card, client_config=config)
# Creates a new text message to be sent to the A2A Server.
# Ex: text_query = 'Why is the sky blue?' # noqa: ERA001
message = new_text_message(text_query, role=Role.ROLE_USER)
request = SendMessageRequest(message=message)
print('Response:')
async for chunk in client.send_message(request):
print(chunk)
create_client 함수는 AgentCard와 ClientConfig가 제공하는 정보를 바탕으로 Client를 만들어요. new_text_message 헬퍼로 Message를 구성하고(role=Role.ROLE_USER 전달), SendMessageRequest로 감싸요. 클라이언트의 send_message 메서드는 에이전트의 단일 최종 Task 또는 Message 응답을 산출하는 async 반복자를 반환해요. 이 예시에서는 Task예요.
클라이언트 초기화 & 스트리밍 메시지 보내기:
print('\nInitializing a streaming client.')
client_config = ClientConfig(streaming=True) # Streaming
client = await create_client(agent=public_agent_card, client_config=client_config)
print('Response:')
async for chunk in client.send_message(request):
print(chunk)
별도의 스트리밍 클라이언트가 ClientConfig에 streaming=True를 넣어 create_client로 생성돼요. 다시 send_message를 호출하는데, 이제 이벤트를 스트리밍해요. 루프 반복마다 네트워크를 통해 도착하는 개별 청크를 출력하죠. 루프 후 await streaming_client.close()를 호출해 기저 HTTP 연결을 해제해요.
예상 출력
test_client.py를 실행하면 다음에 대한 출력이 보여요.
- 공개 agent card가 포맷된 요약으로 표시됨.
- 비스트리밍 응답: 프로토바프 텍스트 형식의 단일
task로, 완료 상태, "Hello, World!"를 담은 생성된 아티팩트, 그리고 이력(history)의 에이전트 중간 상태 메시지를 포함. - 스트리밍 응답: 4개의 청크 — 초기
task, WORKING용status_update, 결과를 담은artifact_update, 그리고 COMPLETED용 최종status_update. - 확장 agent card가 포맷된 요약으로 표시됨(추가
super_hello_world스킬 포함).
출력의 id 필드는 실행마다 달라져요.
AgentCard
--- General ---
Name : Hello World Agent
Description : Just a hello world agent
Version : 0.0.1
--- Interfaces ---
[0] http://127.0.0.1:9999 (JSONRPC)
--- Capabilities ---
Streaming : True
Push notifications : False
Extended agent card : True
--- I/O Modes ---
Input : text/plain
Output : text/plain
--- Skills ---
----------------------------------------------------
ID : hello_world
Name : Returns hello world
Description : just returns hello world
Tags : hello world
Example : hi
Example : hello world
--- Non-Streaming Call ---
Non-streaming Client initialized.
Response:
// Non-streaming response
task {
id: "xxxxxxxx"
context_id: "yyyyyyyy"
status {
state: TASK_STATE_COMPLETED
}
artifacts {
artifact_id: "zzzzzzzz"
name: "result"
parts {
text: "Hello, World!"
}
}
history {
message_id: "vvvvvvvv"
context_id: "yyyyyyyy"
task_id: "xxxxxxxx"
role: ROLE_USER
parts {
text: "Say hello."
}
}
history {
message_id: "wwwwwwww"
role: ROLE_AGENT
parts {
text: "Processing request..."
}
}
}
// Streaming response
task {
id: "xxxxxxxx-s"
context_id: "yyyyyyyy-s"
status {
state: TASK_STATE_SUBMITTED
}
history {
message_id: "vvvvvvvv"
context_id: "yyyyyyyy-s"
task_id: "xxxxxxxx-s"
role: ROLE_USER
parts {
text: "Say hello."
}
}
}
Response chunk:
status_update {
task_id: "xxxxxxxx-s"
context_id: "yyyyyyyy-s"
status {
state: TASK_STATE_WORKING
message {
message_id: "zzzzzzzz-s"
role: ROLE_AGENT
parts {
text: "Processing request..."
}
}
}
}
Response chunk:
artifact_update {
task_id: "xxxxxxxx-s"
context_id: "yyyyyyyy-s"
artifact {
artifact_id: "wwwwwwww-s"
name: "result"
parts {
text: "Hello, World!"
}
}
}
Response chunk:
status_update {
task_id: "xxxxxxxx-s"
context_id: "yyyyyyyy-s"
status {
state: TASK_STATE_COMPLETED
}
}
AgentCard
--- General ---
Name : Hello World Agent - Extended Edition
Description : The full-featured hello world agent for authenticated users.
Version : 0.0.2
--- Interfaces ---
[0] http://127.0.0.1:9999 (JSONRPC)
--- Capabilities ---
Streaming : True
Push notifications : False
Extended agent card : True
--- I/O Modes ---
Input : text/plain
Output : text/plain
--- Skills ---
----------------------------------------------------
ID : hello_world
Name : Returns hello world
Description : just returns hello world
Tags : hello world
Example : hi
Example : hello world
----------------------------------------------------
ID : super_hello_world
Name : Returns a SUPER Hello World
Description : A more enthusiastic greeting, only for authenticated users.
Tags : hello world, super, extended
Example : super hi
Example : give me a super hello
(실제 ID인 xxxxxxxx, yyyyyyyy, zzzzzzzz, wwwwwwww, vvvvvvvv는 실행마다 다른 UUID가 돼요.) 이는 서버가 업데이트된 SDK 구조로 기본 A2A 상호작용을 올바르게 처리하고 있음을 확인해 줘요. 이제 __main__.py가 실행 중인 터미널 창에서 Ctrl+C를 눌러 서버를 종료할 수 있어요.
더 알아보기 (Learn more)
- 스트리밍과 다중 턴을 다루는 다음 단계는 7. Streaming & Multi-Turn Interactions를 참고하세요.
- 클라이언트가 서버와 주고받는 이벤트는 태스크의 수명주기(Life of a Task)에서 확인하세요.