[MCP&A2A] 11. A2A 서버 개발

Agent-to-Agent Protocol (A2A) 서버

A2A 서버는 AI 에이전트 간 비동기 협업을 가능하게 하는 인프라입니다. MCP가 동기식 도구 실행에 집중한다면, A2A는 장기 실행 태스크와 에이전트 간 통신에 최적화되어 있습니다.

A2A vs MCP 비교

특성	A2A	MCP
통신 패턴	비동기 (Task 기반)	동기 (Request-Response)
실행 시간	초~분~시간	밀리초~초
상태 관리	Stateful (태스크 추적)	Stateless
진행 상황	SSE 스트리밍	없음
용도	에이전트 협업, 워크플로우	도구 실행, 데이터 조회
복잡도	높음	낮음

A2A가 필요한 시나리오

✅ A2A 사용:
- 멀티스텝 리서치 (검색 → 분석 → 보고서 생성)
- 코드 리뷰 (분석 → 개선 제안 → 리팩토링)
- 데이터 파이프라인 (수집 → 변환 → 저장)
- 에이전트 팀 협업 (조율자 → 작업자 → 검증자)

❌ A2A 불필요 (MCP 사용):
- 단일 문서 검색
- 간단한 계산
- 빠른 API 호출

프로젝트 구조

a2a-server/
├── cmd/
│   └── server/
│       └── main.go                 # 진입점
├── internal/
│   ├── models/
│   │   ├── task.go                 # 태스크 모델
│   │   ├── agent_card.go           # 에이전트 카드
│   │   └── artifact.go             # 결과물
│   ├── store/
│   │   ├── task_store.go           # 태스크 저장소
│   │   └── postgres_store.go       # PostgreSQL 구현
│   ├── executor/
│   │   ├── executor.go             # 태스크 실행기
│   │   └── worker_pool.go          # 워커 풀
│   ├── handlers/
│   │   ├── agent_card.go           # Agent Card 핸들러
│   │   ├── tasks.go                # 태스크 CRUD
│   │   ├── messages.go             # 메시지 핸들러
│   │   └── events.go               # SSE 스트리밍
│   ├── middleware/
│   │   ├── auth.go                 # JWT 인증
│   │   └── cors.go                 # CORS
│   └── skills/
│       ├── skill.go                # Skill 인터페이스
│       ├── research.go             # 리서치 스킬
│       └── analysis.go             # 분석 스킬
├── go.mod
└── README.md

핵심 모델

1. Task 모델

// internal/models/task.go
package models

import (
    "time"
    
    "github.com/google/uuid"
)

// TaskStatus 열거형
type TaskStatus string

const (
    StatusPending   TaskStatus = "pending"   // 대기 중
    StatusRunning   TaskStatus = "running"   // 실행 중
    StatusCompleted TaskStatus = "completed" // 완료
    StatusFailed    TaskStatus = "failed"    // 실패
    StatusCancelled TaskStatus = "cancelled" // 취소됨
)

// Task 구조체
type Task struct {
    ID        string                 `json:"taskId"`
    AgentID   string                 `json:"agentId"`
    Skill     string                 `json:"skill"`
    Input     map[string]interface{} `json:"input"`
    Status    TaskStatus             `json:"status"`
    Artifacts []Artifact             `json:"artifacts,omitempty"`
    Error     *TaskError             `json:"error,omitempty"`
    
    // 메타데이터
    CreatedAt time.Time `json:"createdAt"`
    UpdatedAt time.Time `json:"updatedAt"`
    StartedAt *time.Time `json:"startedAt,omitempty"`
    EndedAt   *time.Time `json:"endedAt,omitempty"`
    
    // 진행 상황
    Progress  *TaskProgress `json:"progress,omitempty"`
    
    // 내부 필드
    TenantID  string `json:"-"`
    UserID    string `json:"-"`
}

// TaskProgress 진행 상황
type TaskProgress struct {
    Current int    `json:"current"`
    Total   int    `json:"total"`
    Message string `json:"message"`
}

// TaskError 에러 정보
type TaskError struct {
    Code    string `json:"code"`
    Message string `json:"message"`
    Details map[string]interface{} `json:"details,omitempty"`
}

// NewTask 생성자
func NewTask(agentID, skill string, input map[string]interface{}) *Task {
    now := time.Now()
    return &Task{
        ID:        uuid.New().String(),
        AgentID:   agentID,
        Skill:     skill,
        Input:     input,
        Status:    StatusPending,
        CreatedAt: now,
        UpdatedAt: now,
    }
}

// Start 태스크 시작
func (t *Task) Start() {
    now := time.Now()
    t.Status = StatusRunning
    t.StartedAt = &now
    t.UpdatedAt = now
}

// Complete 태스크 완료
func (t *Task) Complete(artifacts []Artifact) {
    now := time.Now()
    t.Status = StatusCompleted
    t.Artifacts = artifacts
    t.EndedAt = &now
    t.UpdatedAt = now
}

// Fail 태스크 실패
func (t *Task) Fail(err error) {
    now := time.Now()
    t.Status = StatusFailed
    t.Error = &TaskError{
        Code:    "EXECUTION_ERROR",
        Message: err.Error(),
    }
    t.EndedAt = &now
    t.UpdatedAt = now
}

// UpdateProgress 진행 상황 업데이트
func (t *Task) UpdateProgress(current, total int, message string) {
    t.Progress = &TaskProgress{
        Current: current,
        Total:   total,
        Message: message,
    }
    t.UpdatedAt = time.Now()
}

// Duration 실행 시간 (완료된 태스크만)
func (t *Task) Duration() time.Duration {
    if t.StartedAt == nil || t.EndedAt == nil {
        return 0
    }
    return t.EndedAt.Sub(*t.StartedAt)
}

2. Artifact 모델

// internal/models/artifact.go
package models

import (
    "time"
    
    "github.com/google/uuid"
)

// Artifact 결과물
type Artifact struct {
    ID        string    `json:"artifactId"`
    Name      string    `json:"name"`
    Parts     []Part    `json:"parts"`
    CreatedAt time.Time `json:"createdAt"`
}

// Part 콘텐츠 블록
type Part struct {
    Kind     string                 `json:"kind"` // text, data, file, image
    Text     string                 `json:"text,omitempty"`
    Data     interface{}            `json:"data,omitempty"`
    URL      string                 `json:"url,omitempty"`
    MimeType string                 `json:"mimeType,omitempty"`
    Metadata map[string]interface{} `json:"metadata,omitempty"`
}

// NewArtifact 생성자
func NewArtifact(name string, parts []Part) *Artifact {
    return &Artifact{
        ID:        uuid.New().String(),
        Name:      name,
        Parts:     parts,
        CreatedAt: time.Now(),
    }
}

// NewTextArtifact 텍스트 결과물
func NewTextArtifact(name, text string) *Artifact {
    return NewArtifact(name, []Part{
        {
            Kind: "text",
            Text: text,
        },
    })
}

// NewDataArtifact 데이터 결과물
func NewDataArtifact(name string, data interface{}) *Artifact {
    return NewArtifact(name, []Part{
        {
            Kind: "data",
            Data: data,
        },
    })
}

// NewFileArtifact 파일 결과물
func NewFileArtifact(name, url, mimeType string) *Artifact {
    return NewArtifact(name, []Part{
        {
            Kind:     "file",
            URL:      url,
            MimeType: mimeType,
        },
    })
}

3. Agent Card 모델

// internal/models/agent_card.go
package models

// AgentCard 에이전트 정의
type AgentCard struct {
    Name           string              `json:"name"`
    Description    string              `json:"description"`
    URL            string              `json:"url"`
    Version        string              `json:"version"`
    Skills         []Skill             `json:"skills"`
    Authentication *AuthenticationInfo `json:"authentication,omitempty"`
}

// Skill 스킬 정의
type Skill struct {
    Name        string   `json:"name"`
    Description string   `json:"description"`
    InputModes  []string `json:"inputModes"`  // text, data, file, image
    OutputModes []string `json:"outputModes"`
}

// AuthenticationInfo 인증 정보
type AuthenticationInfo struct {
    Type        string `json:"type"` // bearer, basic, oauth2
    Description string `json:"description,omitempty"`
}

// NewAgentCard 생성자
func NewAgentCard(name, description, url, version string) *AgentCard {
    return &AgentCard{
        Name:        name,
        Description: description,
        URL:         url,
        Version:     version,
        Skills:      []Skill{},
    }
}

// AddSkill 스킬 추가
func (ac *AgentCard) AddSkill(skill Skill) {
    ac.Skills = append(ac.Skills, skill)
}

Task Store 구현

인터페이스

// internal/store/task_store.go
package store

import (
    "context"
    "time"
    
    "a2a-server/internal/models"
)

// TaskStore 인터페이스
type TaskStore interface {
    // 태스크 생성
    Create(ctx context.Context, task *models.Task) error
    
    // 태스크 조회
    Get(ctx context.Context, taskID string) (*models.Task, error)
    
    // 태스크 목록 (테넌트별)
    List(ctx context.Context, tenantID string, params ListParams) ([]*models.Task, error)
    
    // 태스크 업데이트
    Update(ctx context.Context, task *models.Task) error
    
    // 태스크 삭제
    Delete(ctx context.Context, taskID string) error
    
    // 상태별 조회
    GetByStatus(ctx context.Context, status models.TaskStatus, limit int) ([]*models.Task, error)
}

// ListParams 목록 조회 파라미터
type ListParams struct {
    Status    models.TaskStatus
    AgentID   string
    Limit     int
    Offset    int
    SortBy    string // created_at, updated_at
    SortOrder string // asc, desc
}

PostgreSQL 구현

// internal/store/postgres_store.go
package store

import (
    "context"
    "encoding/json"
    "fmt"
    
    "github.com/jackc/pgx/v5/pgxpool"
    
    "a2a-server/internal/models"
)

type PostgresStore struct {
    pool *pgxpool.Pool
}

func NewPostgresStore(pool *pgxpool.Pool) *PostgresStore {
    return &PostgresStore{pool: pool}
}

// Create 태스크 생성
func (s *PostgresStore) Create(ctx context.Context, task *models.Task) error {
    inputJSON, err := json.Marshal(task.Input)
    if err != nil {
        return fmt.Errorf("failed to marshal input: %w", err)
    }
    
    query := `
        INSERT INTO tasks (
            id, tenant_id, user_id, agent_id, skill,
            input, status, created_at, updated_at
        ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
    `
    
    _, err = s.pool.Exec(
        ctx, query,
        task.ID,
        task.TenantID,
        task.UserID,
        task.AgentID,
        task.Skill,
        inputJSON,
        task.Status,
        task.CreatedAt,
        task.UpdatedAt,
    )
    
    return err
}

// Get 태스크 조회
func (s *PostgresStore) Get(ctx context.Context, taskID string) (*models.Task, error) {
    query := `
        SELECT 
            id, tenant_id, user_id, agent_id, skill,
            input, status, artifacts, error,
            created_at, updated_at, started_at, ended_at,
            progress
        FROM tasks
        WHERE id = $1
    `
    
    var task models.Task
    var inputJSON, artifactsJSON, errorJSON, progressJSON []byte
    var startedAt, endedAt *time.Time
    
    err := s.pool.QueryRow(ctx, query, taskID).Scan(
        &task.ID,
        &task.TenantID,
        &task.UserID,
        &task.AgentID,
        &task.Skill,
        &inputJSON,
        &task.Status,
        &artifactsJSON,
        &errorJSON,
        &task.CreatedAt,
        &task.UpdatedAt,
        &startedAt,
        &endedAt,
        &progressJSON,
    )
    
    if err != nil {
        return nil, err
    }
    
    // JSON 언마샬링
    if err := json.Unmarshal(inputJSON, &task.Input); err != nil {
        return nil, err
    }
    
    if len(artifactsJSON) > 0 {
        if err := json.Unmarshal(artifactsJSON, &task.Artifacts); err != nil {
            return nil, err
        }
    }
    
    if len(errorJSON) > 0 {
        if err := json.Unmarshal(errorJSON, &task.Error); err != nil {
            return nil, err
        }
    }
    
    if len(progressJSON) > 0 {
        if err := json.Unmarshal(progressJSON, &task.Progress); err != nil {
            return nil, err
        }
    }
    
    task.StartedAt = startedAt
    task.EndedAt = endedAt
    
    return &task, nil
}

// Update 태스크 업데이트
func (s *PostgresStore) Update(ctx context.Context, task *models.Task) error {
    inputJSON, _ := json.Marshal(task.Input)
    artifactsJSON, _ := json.Marshal(task.Artifacts)
    errorJSON, _ := json.Marshal(task.Error)
    progressJSON, _ := json.Marshal(task.Progress)
    
    query := `
        UPDATE tasks SET
            status = $2,
            input = $3,
            artifacts = $4,
            error = $5,
            updated_at = $6,
            started_at = $7,
            ended_at = $8,
            progress = $9
        WHERE id = $1
    `
    
    _, err := s.pool.Exec(
        ctx, query,
        task.ID,
        task.Status,
        inputJSON,
        artifactsJSON,
        errorJSON,
        task.UpdatedAt,
        task.StartedAt,
        task.EndedAt,
        progressJSON,
    )
    
    return err
}

// List 태스크 목록
func (s *PostgresStore) List(ctx context.Context, tenantID string, params ListParams) ([]*models.Task, error) {
    query := `
        SELECT 
            id, agent_id, skill, status,
            created_at, updated_at
        FROM tasks
        WHERE tenant_id = $1
    `
    
    args := []interface{}{tenantID}
    argIndex := 2
    
    // 필터 추가
    if params.Status != "" {
        query += fmt.Sprintf(" AND status = $%d", argIndex)
        args = append(args, params.Status)
        argIndex++
    }
    
    if params.AgentID != "" {
        query += fmt.Sprintf(" AND agent_id = $%d", argIndex)
        args = append(args, params.AgentID)
        argIndex++
    }
    
    // 정렬
    sortBy := params.SortBy
    if sortBy == "" {
        sortBy = "created_at"
    }
    
    sortOrder := params.SortOrder
    if sortOrder == "" {
        sortOrder = "desc"
    }
    
    query += fmt.Sprintf(" ORDER BY %s %s", sortBy, sortOrder)
    
    // 페이지네이션
    if params.Limit > 0 {
        query += fmt.Sprintf(" LIMIT $%d", argIndex)
        args = append(args, params.Limit)
        argIndex++
    }
    
    if params.Offset > 0 {
        query += fmt.Sprintf(" OFFSET $%d", argIndex)
        args = append(args, params.Offset)
    }
    
    rows, err := s.pool.Query(ctx, query, args...)
    if err != nil {
        return nil, err
    }
    defer rows.Close()
    
    var tasks []*models.Task
    for rows.Next() {
        var task models.Task
        err := rows.Scan(
            &task.ID,
            &task.AgentID,
            &task.Skill,
            &task.Status,
            &task.CreatedAt,
            &task.UpdatedAt,
        )
        if err != nil {
            return nil, err
        }
        
        tasks = append(tasks, &task)
    }
    
    return tasks, nil
}

// GetByStatus 상태별 조회
func (s *PostgresStore) GetByStatus(ctx context.Context, status models.TaskStatus, limit int) ([]*models.Task, error) {
    query := `
        SELECT id, tenant_id, agent_id, skill, input, status, created_at
        FROM tasks
        WHERE status = $1
        ORDER BY created_at ASC
        LIMIT $2
    `
    
    rows, err := s.pool.Query(ctx, query, status, limit)
    if err != nil {
        return nil, err
    }
    defer rows.Close()
    
    var tasks []*models.Task
    for rows.Next() {
        var task models.Task
        var inputJSON []byte
        
        err := rows.Scan(
            &task.ID,
            &task.TenantID,
            &task.AgentID,
            &task.Skill,
            &inputJSON,
            &task.Status,
            &task.CreatedAt,
        )
        if err != nil {
            return nil, err
        }
        
        json.Unmarshal(inputJSON, &task.Input)
        tasks = append(tasks, &task)
    }
    
    return tasks, nil
}

데이터베이스 스키마

-- tasks 테이블
CREATE TABLE tasks (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL,
    user_id TEXT NOT NULL,
    agent_id TEXT NOT NULL,
    skill TEXT NOT NULL,
    
    -- 입력/출력
    input JSONB NOT NULL,
    artifacts JSONB,
    error JSONB,
    
    -- 상태
    status TEXT NOT NULL,
    progress JSONB,
    
    -- 타임스탬프
    created_at TIMESTAMPTZ NOT NULL,
    updated_at TIMESTAMPTZ NOT NULL,
    started_at TIMESTAMPTZ,
    ended_at TIMESTAMPTZ
);

-- 인덱스
CREATE INDEX idx_tasks_tenant_id ON tasks(tenant_id);
CREATE INDEX idx_tasks_status ON tasks(status) WHERE status IN ('pending', 'running');
CREATE INDEX idx_tasks_agent_id ON tasks(agent_id);
CREATE INDEX idx_tasks_created_at ON tasks(created_at DESC);

-- RLS 활성화
ALTER TABLE tasks ENABLE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation ON tasks
    USING (tenant_id = current_setting('app.current_tenant_id', true)::uuid);

Task Executor 구현

Executor 인터페이스

// internal/executor/executor.go
package executor

import (
    "context"
    
    "a2a-server/internal/models"
)

// Executor 태스크 실행기
type Executor interface {
    Execute(ctx context.Context, task *models.Task) error
}

// SkillExecutor 스킬별 실행기
type SkillExecutor interface {
    Name() string
    Execute(ctx context.Context, input map[string]interface{}) ([]models.Artifact, error)
}

Worker Pool 구현

// internal/executor/worker_pool.go
package executor

import (
    "context"
    "log"
    "sync"
    
    "a2a-server/internal/models"
    "a2a-server/internal/store"
)

type WorkerPool struct {
    taskStore store.TaskStore
    skills    map[string]SkillExecutor
    
    workers   int
    taskQueue chan *models.Task
    wg        sync.WaitGroup
    
    eventBus *EventBus
}

func NewWorkerPool(
    taskStore store.TaskStore,
    workers int,
    eventBus *EventBus,
) *WorkerPool {
    return &WorkerPool{
        taskStore: taskStore,
        skills:    make(map[string]SkillExecutor),
        workers:   workers,
        taskQueue: make(chan *models.Task, workers*2),
        eventBus:  eventBus,
    }
}

// RegisterSkill 스킬 등록
func (wp *WorkerPool) RegisterSkill(skill SkillExecutor) {
    wp.skills[skill.Name()] = skill
}

// Start 워커 풀 시작
func (wp *WorkerPool) Start(ctx context.Context) {
    log.Printf("Starting worker pool with %d workers", wp.workers)
    
    for i := 0; i < wp.workers; i++ {
        wp.wg.Add(1)
        go wp.worker(ctx, i)
    }
    
    // 대기 중인 태스크 폴링
    go wp.pollPendingTasks(ctx)
}

// Stop 워커 풀 중지
func (wp *WorkerPool) Stop() {
    close(wp.taskQueue)
    wp.wg.Wait()
    log.Println("Worker pool stopped")
}

// Submit 태스크 제출
func (wp *WorkerPool) Submit(task *models.Task) {
    wp.taskQueue <- task
}

// worker 개별 워커
func (wp *WorkerPool) worker(ctx context.Context, id int) {
    defer wp.wg.Done()
    
    log.Printf("Worker %d started", id)
    
    for task := range wp.taskQueue {
        log.Printf("Worker %d processing task %s", id, task.ID)
        
        if err := wp.executeTask(ctx, task); err != nil {
            log.Printf("Worker %d failed task %s: %v", id, task.ID, err)
        }
    }
    
    log.Printf("Worker %d stopped", id)
}

// executeTask 태스크 실행
func (wp *WorkerPool) executeTask(ctx context.Context, task *models.Task) error {
    // 스킬 찾기
    skill, ok := wp.skills[task.Skill]
    if !ok {
        task.Fail(fmt.Errorf("skill not found: %s", task.Skill))
        wp.taskStore.Update(ctx, task)
        wp.eventBus.Publish(task.ID, Event{
            Type: "error",
            Data: map[string]interface{}{
                "message": "Skill not found",
            },
        })
        return fmt.Errorf("skill not found: %s", task.Skill)
    }
    
    // 태스크 시작
    task.Start()
    wp.taskStore.Update(ctx, task)
    wp.eventBus.Publish(task.ID, Event{
        Type: "started",
        Data: map[string]interface{}{
            "taskId": task.ID,
        },
    })
    
    // 스킬 실행
    artifacts, err := skill.Execute(ctx, task.Input)
    
    if err != nil {
        // 실패 처리
        task.Fail(err)
        wp.taskStore.Update(ctx, task)
        wp.eventBus.Publish(task.ID, Event{
            Type: "error",
            Data: map[string]interface{}{
                "message": err.Error(),
            },
        })
        return err
    }
    
    // 완료 처리
    task.Complete(artifacts)
    wp.taskStore.Update(ctx, task)
    wp.eventBus.Publish(task.ID, Event{
        Type: "completed",
        Data: map[string]interface{}{
            "taskId":    task.ID,
            "artifacts": artifacts,
        },
    })
    
    log.Printf("Task %s completed in %v", task.ID, task.Duration())
    
    return nil
}

// pollPendingTasks 대기 중인 태스크 폴링
func (wp *WorkerPool) pollPendingTasks(ctx context.Context) {
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()
    
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            tasks, err := wp.taskStore.GetByStatus(ctx, models.StatusPending, wp.workers)
            if err != nil {
                log.Printf("Failed to poll pending tasks: %v", err)
                continue
            }
            
            for _, task := range tasks {
                select {
                case wp.taskQueue <- task:
                    log.Printf("Queued pending task %s", task.ID)
                default:
                    log.Printf("Queue full, skipping task %s", task.ID)
                }
            }
        }
    }
}

Event Bus (SSE용)

// internal/executor/event_bus.go
package executor

import (
    "sync"
)

type Event struct {
    Type string
    Data map[string]interface{}
}

type EventBus struct {
    subscribers map[string][]chan Event
    mu          sync.RWMutex
}

func NewEventBus() *EventBus {
    return &EventBus{
        subscribers: make(map[string][]chan Event),
    }
}

// Subscribe 이벤트 구독
func (eb *EventBus) Subscribe(taskID string) chan Event {
    eb.mu.Lock()
    defer eb.mu.Unlock()
    
    ch := make(chan Event, 10)
    eb.subscribers[taskID] = append(eb.subscribers[taskID], ch)
    
    return ch
}

// Unsubscribe 구독 취소
func (eb *EventBus) Unsubscribe(taskID string, ch chan Event) {
    eb.mu.Lock()
    defer eb.mu.Unlock()
    
    subs := eb.subscribers[taskID]
    for i, sub := range subs {
        if sub == ch {
            eb.subscribers[taskID] = append(subs[:i], subs[i+1:]...)
            close(ch)
            break
        }
    }
    
    if len(eb.subscribers[taskID]) == 0 {
        delete(eb.subscribers, taskID)
    }
}

// Publish 이벤트 발행
func (eb *EventBus) Publish(taskID string, event Event) {
    eb.mu.RLock()
    defer eb.mu.RUnlock()
    
    for _, ch := range eb.subscribers[taskID] {
        select {
        case ch <- event:
        default:
            // 채널이 가득 차면 무시
        }
    }
}

HTTP 핸들러

Agent Card 핸들러

// internal/handlers/agent_card.go
package handlers

import (
    "encoding/json"
    "net/http"
    
    "a2a-server/internal/models"
)

type AgentCardHandler struct {
    agentCard *models.AgentCard
}

func NewAgentCardHandler(agentCard *models.AgentCard) *AgentCardHandler {
    return &AgentCardHandler{agentCard: agentCard}
}

// GET /.well-known/agent-card.json
func (h *AgentCardHandler) Get(w http.ResponseWriter, r *http.Request) {
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(h.agentCard)
}

Task 핸들러

// internal/handlers/tasks.go
package handlers

import (
    "encoding/json"
    "net/http"
    
    "github.com/go-chi/chi/v5"
    
    "a2a-server/internal/executor"
    "a2a-server/internal/models"
    "a2a-server/internal/store"
)

type TaskHandler struct {
    taskStore  store.TaskStore
    workerPool *executor.WorkerPool
}

func NewTaskHandler(taskStore store.TaskStore, workerPool *executor.WorkerPool) *TaskHandler {
    return &TaskHandler{
        taskStore:  taskStore,
        workerPool: workerPool,
    }
}

// POST /a2a/{agentId}/tasks
func (h *TaskHandler) Create(w http.ResponseWriter, r *http.Request) {
    agentID := chi.URLParam(r, "agentId")
    
    var req struct {
        Skill string                 `json:"skill"`
        Input map[string]interface{} `json:"input"`
    }
    
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        http.Error(w, "Invalid request", http.StatusBadRequest)
        return
    }
    
    // 태스크 생성
    task := models.NewTask(agentID, req.Skill, req.Input)
    task.TenantID = r.Context().Value("tenant_id").(string)
    task.UserID = r.Context().Value("user_id").(string)
    
    // 저장
    if err := h.taskStore.Create(r.Context(), task); err != nil {
        http.Error(w, "Failed to create task", http.StatusInternalServerError)
        return
    }
    
    // 워커 풀에 제출
    h.workerPool.Submit(task)
    
    // 응답
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(http.StatusCreated)
    json.NewEncoder(w).Encode(task)
}

// GET /a2a/{agentId}/tasks/{taskId}
func (h *TaskHandler) Get(w http.ResponseWriter, r *http.Request) {
    taskID := chi.URLParam(r, "taskId")
    
    task, err := h.taskStore.Get(r.Context(), taskID)
    if err != nil {
        http.Error(w, "Task not found", http.StatusNotFound)
        return
    }
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(task)
}

// GET /a2a/{agentId}/tasks
func (h *TaskHandler) List(w http.ResponseWriter, r *http.Request) {
    tenantID := r.Context().Value("tenant_id").(string)
    
    params := store.ListParams{
        Limit:  10,
        Offset: 0,
    }
    
    // 쿼리 파라미터 파싱
    if status := r.URL.Query().Get("status"); status != "" {
        params.Status = models.TaskStatus(status)
    }
    
    tasks, err := h.taskStore.List(r.Context(), tenantID, params)
    if err != nil {
        http.Error(w, "Failed to list tasks", http.StatusInternalServerError)
        return
    }
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(map[string]interface{}{
        "tasks": tasks,
    })
}

SSE 이벤트 핸들러

// internal/handlers/events.go
package handlers

import (
    "encoding/json"
    "fmt"
    "net/http"
    
    "github.com/go-chi/chi/v5"
    
    "a2a-server/internal/executor"
)

type EventHandler struct {
    eventBus *executor.EventBus
}

func NewEventHandler(eventBus *executor.EventBus) *EventHandler {
    return &EventHandler{eventBus: eventBus}
}

// GET /a2a/{agentId}/tasks/{taskId}/events
func (h *EventHandler) Stream(w http.ResponseWriter, r *http.Request) {
    taskID := chi.URLParam(r, "taskId")
    
    // SSE 헤더 설정
    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("Connection", "keep-alive")
    w.Header().Set("Access-Control-Allow-Origin", "*")
    
    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "Streaming not supported", http.StatusInternalServerError)
        return
    }
    
    // 이벤트 구독
    eventChan := h.eventBus.Subscribe(taskID)
    defer h.eventBus.Unsubscribe(taskID, eventChan)
    
    // 이벤트 스트리밍
    for {
        select {
        case event, ok := <-eventChan:
            if !ok {
                return
            }
            
            // JSON 직렬화
            data, _ := json.Marshal(event.Data)
            
            // SSE 포맷으로 전송
            fmt.Fprintf(w, "event: %s\n", event.Type)
            fmt.Fprintf(w, "data: %s\n\n", data)
            flusher.Flush()
            
            // 완료 또는 에러 시 종료
            if event.Type == "completed" || event.Type == "error" {
                return
            }
            
        case <-r.Context().Done():
            return
        }
    }
}

main.go - 전체 조합

// cmd/server/main.go
package main

import (
    "context"
    "log"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"
    
    "github.com/go-chi/chi/v5"
    "github.com/go-chi/chi/v5/middleware"
    "github.com/jackc/pgx/v5/pgxpool"
    
    "a2a-server/internal/executor"
    "a2a-server/internal/handlers"
    custommw "a2a-server/internal/middleware"
    "a2a-server/internal/models"
    "a2a-server/internal/skills"
    "a2a-server/internal/store"
)

func main() {
    // 환경 변수
    dbURL := os.Getenv("DATABASE_URL")
    port := os.Getenv("PORT")
    if port == "" {
        port = "8081"
    }
    
    // DB 연결
    pool, err := pgxpool.New(context.Background(), dbURL)
    if err != nil {
        log.Fatal(err)
    }
    defer pool.Close()
    
    log.Println("✅ Database connected")
    
    // Store 생성
    taskStore := store.NewPostgresStore(pool)
    
    // Event Bus
    eventBus := executor.NewEventBus()
    
    // Worker Pool
    workerPool := executor.NewWorkerPool(taskStore, 10, eventBus)
    
    // 스킬 등록
    workerPool.RegisterSkill(skills.NewResearchSkill())
    workerPool.RegisterSkill(skills.NewAnalysisSkill())
    
    log.Println("✅ Skills registered")
    
    // Worker Pool 시작
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    
    workerPool.Start(ctx)
    log.Println("✅ Worker pool started")
    
    // Agent Card
    agentCard := models.NewAgentCard(
        "Research Assistant",
        "Multi-step research and analysis agent",
        "https://api.example.com/a2a/research-assistant",
        "1.0.0",
    )
    agentCard.AddSkill(models.Skill{
        Name:        "research",
        Description: "Comprehensive research on a topic",
        InputModes:  []string{"text"},
        OutputModes: []string{"text", "data"},
    })
    
    // 핸들러
    agentCardHandler := handlers.NewAgentCardHandler(agentCard)
    taskHandler := handlers.NewTaskHandler(taskStore, workerPool)
    eventHandler := handlers.NewEventHandler(eventBus)
    
    // 라우터
    r := chi.NewRouter()
    
    // 미들웨어
    r.Use(middleware.RequestID)
    r.Use(middleware.RealIP)
    r.Use(middleware.Logger)
    r.Use(middleware.Recoverer)
    
    // Agent Card (공개)
    r.Get("/.well-known/agent-card.json", agentCardHandler.Get)
    
    // A2A 엔드포인트 (인증 필요)
    r.Group(func(r chi.Router) {
        r.Use(custommw.AuthMiddleware(jwtValidator))
        
        r.Post("/a2a/{agentId}/tasks", taskHandler.Create)
        r.Get("/a2a/{agentId}/tasks/{taskId}", taskHandler.Get)
        r.Get("/a2a/{agentId}/tasks", taskHandler.List)
        r.Get("/a2a/{agentId}/tasks/{taskId}/events", eventHandler.Stream)
    })
    
    // 서버 시작
    srv := &http.Server{
        Addr:    ":" + port,
        Handler: r,
    }
    
    go func() {
        log.Printf("🚀 A2A Server listening on :%s", port)
        if err := srv.ListenAndServe(); err != http.ErrServerClosed {
            log.Fatal(err)
        }
    }()
    
    // Graceful shutdown
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    <-quit
    
    log.Println("Shutting down server...")
    
    cancel() // Worker pool 중지
    workerPool.Stop()
    
    shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer shutdownCancel()
    
    if err := srv.Shutdown(shutdownCtx); err != nil {
        log.Fatal(err)
    }
    
    log.Println("✅ Server stopped")
}

핵심 요약

A2A 서버 구성

• ✅ Models: Task, Artifact, AgentCard
• ✅ Store: PostgreSQL 기반 영속성
• ✅ Executor: Worker Pool + Event Bus
• ✅ Handlers: Task CRUD + SSE 스트리밍
• ✅ Skills: 비즈니스 로직 구현

주요 특징

• ✅ 비동기: 백그라운드 Worker Pool
• ✅ 확장성: 워커 수 조정 가능
• ✅ 실시간: SSE로 진행 상황 스트리밍
• ✅ 상태 관리: PostgreSQL에 태스크 저장
• ✅ 멀티테넌시: RLS로 격리

작성일: 2024-12-13