IT 헬프데스크 서비스를 만든다고 해 봅니다. 직원이 "VPN이 안 돼요"라고 물으면 사내 문서에서 답을 찾아 주고, "T-1042 상태 알려 줘"라고 하면 티켓 시스템을 조회하고, "토너 교체 티켓 만들어 줘"라고 하면 사람의 승인을 받은 뒤에 티켓을 만듭니다. 지금까지 하나씩 만든 부품이 모두 들어갑니다.
부품은 지식 검색 도구, MCP로 연결한 티켓 도구, 쓰기 앞에서 멈추는 사람 승인, 멈춘 자리를 기억하는 저장소입니다. 이 넷을 Agent 하나로 묶고 웹 서버가 바깥에 내놓습니다. 질문을 골라 어떤 부품이 켜지는지 보세요.
설명을 위한 구조도입니다. 세 질문 모두 FastAPI 서비스를 실제로 띄워 스트림으로 확인했습니다. 모델은 모의 서버, 사내 문서와 티켓은 가짜 데이터입니다.
티켓을 만드는 요청은 읽기 요청과 달리 왕복이 두 번입니다. 처음 요청은 Agent가 create_ticket을 부르기로 하는 데서 멈추고 서버가 승인 요청을 스트림으로 내보낸 뒤 연결을 닫습니다. 사람이 답하면 두 번째 요청이 같은 대화 번호(thread_id)로 이어 갑니다.
저장소가 멈춘 자리를 기억하기 때문에 이어 갈 수 있습니다. 스크롤하며 요청이 오가는 순서를 따라가 보세요.
POST /chat에 대화 번호와 질문을 보내면 웹 서버가 Agent를 실행하며 진행 상황을 스트림으로 흘려 보냅니다.
Agent가 create_ticket을 부르려는 순간 승인 미들웨어가 멈춥니다. 스트림에 approval_required 이벤트가 나가고 응답은 끝납니다.
화면이 보낼 내용을 보여 주면 사람이 고릅니다. 같은 대화 번호로 POST /approve를 보냅니다.
Agent가 저장소의 기록으로 이어서 MCP 서버에 티켓을 만들고 결과를 답으로 흘려 보냅니다. 거절하면 도구는 실행되지 않습니다.
Agent는 도구를 몇 번 부를지 모르고 몇 초가 걸릴지도 모릅니다. 응답이 끝나기를 기다렸다가 한 번에 보내면 사용자는 그동안 빈 화면만 보게 됩니다. SSE(Server-Sent Events)는 서버가 연결을 열어 둔 채 글 조각을 계속 보내는 방식입니다.
형식은 단순합니다. 이벤트 이름 줄(event:)과 내용 줄(data:)을 쓰고 빈 줄로 하나를 끝냅니다. 응답의 종류(media_type)는 text/event-stream으로 정합니다. 방식을 바꿔 보세요.
서버가 켜질 때(lifespan) 부품을 한 번만 만들어 둡니다. 검색 도구와 MCP 서버에서 받은 도구를 합쳐 Agent에게 주고 쓰기 도구에만 승인을 걸고 저장소를 붙입니다. 요청이 올 때마다 Agent를 새로 만들지 않고 만들어 둔 것을 씁니다.
@asynccontextmanager
async def lifespan(app):
tools = [search_kb, *await mcp.get_tools()] # 1 검색 도구 + MCP 도구
app.state.agent = create_agent(model, tools=tools,
middleware=[HumanInTheLoopMiddleware(
interrupt_on={"create_ticket": True})], # 2 쓰기만 승인
checkpointer=InMemorySaver()) # 3 멈춘 자리 기억
yieldcreate_ticket은 사람의 답을 기다립니다. 읽기와 쓰기를 나눠 불편은 줄이고 위험은 막습니다.InMemorySaver)라 서버를 다시 켜면 대화가 사라집니다. 실제 서비스는 데이터베이스용 체크포인터를 씁니다.async def run(agent, inputs, thread_id):
async for chunk in agent.astream(inputs, cfg, stream_mode="updates"): # 1 단계가 끝날 때마다
for node, upd in chunk.items():
if node == "__interrupt__": # 2 승인 대기
req = upd[0].value["action_requests"][0]
yield sse("approval_required", {"tool": req["name"], "args": req["args"]}); return
... # tool_call·tool_result·answer
yield sse("done", {}) # 3 끝출력은 예시입니다. FastAPI 서비스를 띄워 curl로 직접 확인한 스트림을 줄였고 모델은 모의 서버입니다. 같은 서비스에서 지식 질문과 조회 질문, 거절 경로도 확인했습니다.
stream_mode="updates"는 노드가 끝날 때마다 바뀐 것만 줍니다. 그 안에서 도구 호출, 도구 결과, 답을 골라 SSE 이벤트로 바꿉니다.__interrupt__가 담겨 옵니다. 보내려는 도구와 인자를 이벤트로 흘리고 이번 응답은 끝냅니다.done 이벤트를 보내 화면이 입력창을 다시 열 수 있게 합니다.마지막 글에서는 이 서비스를 컨테이너로 묶어 남에게 넘기는 방법과 평가 결과, 설계 기록을 정리하겠습니다.