A2A 샘플: Travel Planner OpenRouter

Cheng Zhang·2025년 6월 7일

이것은 A2A(Agent2Agent) 프로토콜을 준수하는 Python 구현입니다.
Google의 공식 a2a-python SDK와 OpenAI Python SDK를 기반으로 구현된 여행 어시스턴트 데모입니다.
OpenAI 모델 사양에 맞는 여행 어시스턴트로, 여행 계획 서비스를 제공할 수 있습니다.

소스 코드

A2A Travel Planner OpenRouter

프로젝트 아키텍처

이 프로젝트는 A2A 프로토콜을 사용하여 상호 운용 가능한 여행 계획 에이전트를 구축하는 방법을 보여주며, 다음과 같은 핵심 구성 요소를 포함합니다:

Travel Planner Agent: OpenAI 호환 인터페이스를 기반으로 한 핵심 여행 어시스턴트 로직
Agent Executor: 에이전트 로직을 A2A 서버에 연결하는 A2A 프로토콜 어댑터
A2A Server: 표준화된 에이전트 간 통신 인터페이스를 제공하는 A2A 프로토콜 준수 서버
Loop Client: A2A 서버와 상호 작용하기 위한 테스트 클라이언트

워크플로우 시퀀스 다이어그램

sequenceDiagram
    participant Client
    participant A2AServer
    participant RequestHandler
    participant Executor as TravelPlannerAgentExecutor
    participant Agent as TravelPlannerAgent
    participant LLM as OpenAI-Compatible LLM

    Client->>A2AServer: 에이전트 카드 요청
    A2AServer->>Client: 에이전트 카드 반환 (기술, 능력)

    Note over Client,A2AServer: 사용자가 여행 계획 쿼리

    Client->>A2AServer: message/sendStream (스트리밍 요청)
    A2AServer->>RequestHandler: 스트리밍 요청 라우팅
    RequestHandler->>Executor: execute(context, event_queue)
    
    Executor->>Agent: stream(query)
    Agent->>LLM: chat.completions.create(stream=True)
    
    loop 스트리밍 응답 처리
        LLM-->>Agent: 스트리밍 콘텐츠 청크 반환
        Agent-->>Executor: yield {'content': chunk, 'done': False}
        Executor-->>RequestHandler: TaskArtifactUpdateEvent
        RequestHandler-->>A2AServer: SSE 이벤트 푸시
        A2AServer-->>Client: 스트림 콘텐츠 업데이트
    end
    
    LLM-->>Agent: 최종 응답 완료
    Agent-->>Executor: yield {'content': '', 'done': True}
    Executor-->>RequestHandler: 최종 TaskArtifactUpdateEvent
    RequestHandler-->>A2AServer: 최종 SSE 이벤트
    A2AServer-->>Client: 스트리밍 응답 완료

주요 워크플로우

에이전트 카드 검색: 클라이언트는 먼저 A2A 서버에서 에이전트 카드를 검색하여 에이전트의 능력과 기술을 이해합니다
스트리밍 요청 처리: 클라이언트는 사용자 쿼리가 포함된 스트리밍 메시지 요청을 보냅니다
에이전트 실행: 에이전트 실행기가 요청을 처리하고 여행 계획 에이전트 핵심 로직을 호출합니다
LLM 상호작용: 에이전트는 OpenAI 호환 LLM과 스트리밍 대화를 수행합니다
실시간 응답: Server-Sent Events(SSE)를 통해 클라이언트에 실시간으로 응답을 스트리밍합니다

시작하기

환경 변수를 구성합니다:

예제 파일을 복사하고 API 자격 증명을 구성하세요.

cp env.example .env

실제 값으로 .env 파일을 편집합니다:

# 필수: AI 모델 서비스용 API 키
API_KEY=your_actual_api_key_here

# 선택사항: 모델 이름 (기본값: google/gemini-2.0-flash-001)
MODEL_NAME=google/gemini-2.0-flash-001

# 선택사항: API 서비스의 기본 URL
BASE_URL=https://openrouter.ai/api/v1

종속성을 설치하고 서버를 시작합니다:

uv venv
source .venv/bin/activate

uv sync
uv run .

새 터미널에서 루프 클라이언트를 실행합니다:
```
source .venv/bin/activate
uv run loop_client.py
```

구성

애플리케이션은 구성을 위해 환경 변수를 사용합니다:

API_KEY (필수): AI 모델 서비스용 API 키
MODEL_NAME (선택사항): 사용할 모델 이름 (기본값: "google/gemini-2.0-flash-001")
BASE_URL (선택사항): API 서비스의 기본 URL (기본값: "https://openrouter.ai/api/v1")

기술적 특징

현재 구현

✅ A2A 프로토콜 준수: Agent2Agent 프로토콜 사양에 완전히 준수
✅ 스트리밍 응답: 실시간 스트리밍 콘텐츠 생성 지원
✅ OpenAI 호환: 모든 OpenAI 호환 API 인터페이스 지원
✅ 모듈러 설계: 에이전트 로직과 프로토콜 적응의 명확한 분리
✅ 환경 구성: 유연한 환경 변수 구성

향후 개선 계획

작업 상태 관리 개선

Google A2A LangGraph 샘플을 기반으로 다음 기능을 추가할 계획입니다:

🔄 작업 생명주기 관리: 완전한 작업 상태 추적 구현 (제출됨 → 작업 중 → 완료/실패)
🔄 다중 턴 대화 지원: 사용자 명확화가 필요한 복잡한 여행 계획 시나리오를 지원하는 input_required 상태 추가
🔄 작업 지속성: 장기 실행 계획 작업을 위한 작업 상태 지속성 구현
🔄 향상된 오류 처리: 더 자세한 오류 상태 및 복구 메커니즘
🔄 작업 취소: 진행 중인 작업의 취소 지원

상태 관리 예제

# 향후 상태 관리 구현 예제
class TravelPlannerTaskManager:
    async def handle_complex_query(self, query: str, context: RequestContext):
        # 더 많은 정보가 필요한지 감지
        if self.needs_clarification(query):
            return TaskStatus(
                state=TaskState.input_required,
                message="더 많은 정보가 필요합니다: 구체적인 목적지, 날짜 및 예산 범위를 제공해 주세요"
            )
        
        # 복잡한 다단계 계획 실행
        task_id = await self.create_long_running_task(query)
        return TaskStatus(
            state=TaskState.working,
            taskId=task_id,
            message="상세한 여행 계획을 생성 중..."
        )