메시지 패싱 기초

의 본질은 결국 를 어떻게 흘려보내느냐다. 본격적으로 스펙을 파기 전에 ·· 같은 기초 어휘를 통일하자.

한 줄 정의로 시작

에이전트끼리 무엇을 주고받는가? 답은 단순하다 — 한 덩어리다.

함수 호출과 다른 점은 호출자와 수신자가 같은 프로세스에 있을 필요가 없다는 것. 그래서 라는 중간 인프라가 등장하고, 과 이라는 두 갈래가 갈린다.

메시지는 “리모트 함수 호출의 일반화”가 아니다. 그 반대로, 함수 호출이 “동기 메시지의 특수 케이스”다.

동기 vs 비동기: 무엇이 다른가

같은 를 보내도 호출자가 기다리느냐에 따라 모든 트레이드오프가 달라진다.

축
지연	가장 느린 단계에 종속	호출자 즉시 반환
실패 모드	타임아웃 → 호출자가 책임	메시지 유실·중복 가능
코드 복잡도	낮다	콜백·폴링·재시도 필요
좋은 곳	짧은 한 턴	장시간 작업

A2A는 두 모드를 모두 지원한다. message/send는 동기, message/stream은 SSE 비동기, 그리고 push notification은 완전 비동기다.

점대점: 가장 단순한 모델

다이어그램 로딩…

한 송신자가 정확히 한 수신자에게 를 꽂아 넣는다. 은 보통 한 개로 구현한다. 워커가 여러 명이어도 한 메시지는 한 만 집어 간다.

장점: 흐름이 단순하고 처리 보장이 명확하다. 단점: 송신자가 수신자의 주소(엔드포인트·큐 이름)를 알아야 한다.

브로드캐스트와 Pub/Sub

한 메시지를 여러 명이 봐야 한다면 모델이 달라진다.

: “지금 듣고 있는 모두에게” 한 번에 뿌린다. 보통 휘발성.
: 발행자는 토픽으로만 던지고, 구독자는 자신이 관심 있는 토픽만 받아 간다. 이 곧 토픽이다.

다이어그램 로딩…

핵심은 디커플링이다. 발행자는 누가 듣는지 모르고, 구독자는 누가 보냈는지 신경 쓸 필요가 없다.

Python asyncio.Queue로 동기·비동기 비교

표준 라이브러리 만으로 를 만든다. 생산자가 를 넣고, 소비자가 빼낸다. 이게 의 가장 작은 단위다.

# Verified against: https://docs.python.org/3/library/asyncio-queue.html
# Verified at: 2026-06-02
import asyncio

async def producer(q: asyncio.Queue):
  for i in range(3):
      await q.put({"id": i, "text": f"msg-{i}"})
  await q.put(None)  # 종료 신호

async def consumer(q: asyncio.Queue):
  while True:
      m = await q.get()
      if m is None:
          break
      print("got", m)

async def main():
  q: asyncio.Queue = asyncio.Queue()
  await asyncio.gather(producer(q), consumer(q))

asyncio.run(main())

// Verified against: https://nodejs.org/api/events.html (EventEmitter)
// Verified at: 2026-06-02
type Msg = { id: number; text: string } | null

class Channel<T> {
private waiters: Array<(v: T) => void> = []
private buffer: T[] = []
put(v: T) {
  const w = this.waiters.shift()
  if (w) w(v)
  else this.buffer.push(v)
}
get(): Promise<T> {
  const v = this.buffer.shift()
  if (v !== undefined) return Promise.resolve(v)
  return new Promise((res) => this.waiters.push(res))
}
}

async function main() {
const q = new Channel<Msg>()
;(async () => {
  for (let i = 0; i < 3; i++) q.put({ id: i, text: `msg-${i}` })
  q.put(null)
})()
while (true) {
  const m = await q.get()
  if (m === null) break
  console.log('got', m)
}
}

main()

큐 한 개 = 점대점

같은 에 소비자를 여럿 붙이면? 의 풀이 된다. 한 는 어느 한 워커가 가져가지만, 어떤 워커냐는 보장 없다. 자연스러운 부하 분산 효과가 생긴다.

# Verified against: https://docs.python.org/3/library/asyncio-queue.html
# Verified at: 2026-06-02
import asyncio

async def worker(name: str, q: asyncio.Queue):
  while True:
      m = await q.get()
      if m is None:
          q.task_done()
          break
      print(name, "->", m)
      q.task_done()

async def main():
  q: asyncio.Queue = asyncio.Queue()
  workers = [asyncio.create_task(worker(f"w{i}", q)) for i in range(3)]
  for i in range(10):
      await q.put(i)
  for _ in workers:
      await q.put(None)
  await asyncio.gather(*workers)

asyncio.run(main())

// Verified against: https://nodejs.org/api/async_hooks.html
// Verified at: 2026-06-02
async function worker(name: string, q: AsyncIterable<number | null>) {
for await (const m of q) {
  if (m === null) return
  console.log(name, '->', m)
}
}

function makeQueue<T>() {
const buf: T[] = []
const waiters: Array<(v: T) => void> = []
return {
  put(v: T) { const w = waiters.shift(); w ? w(v) : buf.push(v) },
  async *[Symbol.asyncIterator]() {
    while (true) {
      yield await new Promise<T>((r) => {
        const v = buf.shift(); v !== undefined ? r(v) : waiters.push(r)
      })
    }
  },
}
}

Pub/Sub: 한 채널, 다 구독자

같은 를 모든 구독자에게 복제해 보낸다. 의 핵심이다. 구독자가 늦게 들어오면 그 이전 메시지는 못 볼 수도 있다(휘발성). 보존이 필요하면 가 로그를 유지해야 한다(Kafka, Redis Streams).

# Verified against: https://docs.python.org/3/library/asyncio-queue.html
# Verified at: 2026-06-02
import asyncio
from collections import defaultdict

class PubSub:
  def __init__(self):
      self._subs: dict[str, list[asyncio.Queue]] = defaultdict(list)

  def subscribe(self, topic: str) -> asyncio.Queue:
      q: asyncio.Queue = asyncio.Queue()
      self._subs[topic].append(q)
      return q

  async def publish(self, topic: str, msg):
      for q in self._subs[topic]:
          await q.put(msg)

async def main():
  bus = PubSub()
  a, b = bus.subscribe("task.done"), bus.subscribe("task.done")
  await bus.publish("task.done", {"id": 1})
  print(await a.get(), await b.get())

asyncio.run(main())

// Verified against: https://nodejs.org/api/events.html
// Verified at: 2026-06-02
import { EventEmitter } from 'node:events'

class PubSub {
private ee = new EventEmitter()
subscribe<T>(topic: string, fn: (m: T) => void) {
  this.ee.on(topic, fn)
}
publish<T>(topic: string, msg: T) {
  this.ee.emit(topic, msg)
}
}

const bus = new PubSub()
bus.subscribe<{ id: number }>('task.done', (m) => console.log('A', m))
bus.subscribe<{ id: number }>('task.done', (m) => console.log('B', m))
bus.publish('task.done', { id: 1 })

백프레셔와 버퍼

는 무한하지 않다. 생산자가 소비자보다 빠르면 큐가 부풀고, 메모리가 터지거나 를 버려야 한다.

대응 전략 세 가지:

블로킹: Queue(maxsize=N). 가득 차면 생산자가 멈춘다. 자연스러운 백프레셔.
드롭: 오래된 메시지부터 버린다. 모니터링·로그 같은 손실 허용 흐름에 적합.
스필: 디스크·외부 로 흘려보낸다. Redis Streams·Kafka가 이 역할.

시스템에서는 1번 + 3번 조합이 일반적이다. 학습용 인메모리 데모는 1번으로 충분하다.

메시지 버스 한 장으로 보기

다이어그램 로딩…

는 위 세 모델을 모두 품을 수 있는 일반 인프라다. 토픽이 한 개면 , 다수 구독자가 같은 토픽이면 , 토픽이 여럿이면 . 같은 버스를 어떻게 논리적으로 쓰느냐의 차이일 뿐이다.

프로덕션에서는 보통 Redis Streams, Kafka, NATS, RabbitMQ 중 하나가 이 자리를 차지한다. 자세한 비교는 20장에서 다룬다.

어떤 모델을 언제 쓰나

시나리오	추천 모델	이유
한 턴 호출		단순·즉답
장기 작업 (코드 생성·렌더링)	+	호출자 비차단
알림·이벤트 전파		디커플링
워커 풀 분산	+ 다중 소비자	자연스러운 부하 분산

규칙: 응답이 필요하면 점대점, 사실 통지면 Pub/Sub, 작업 위임이면 큐 + 결과 채널.

다음 챕터로

큐와 채널은 그릇이다. 그 안에 흐를 메시지 자체의 모양이 정해져 있지 않으면 양쪽 끝이 서로 다르게 해석한다. 다음 장에서는 메시지의 스키마를 잡는다 — ···까지.