LangGraph와 FastMCP를 사용한 오픈소스 에이전트 워크스페이스 구축 (12개 이상의 도구, 실시간 SSE, 99.99% SLA)
요약
본 기사는 LangGraph와 FastMCP를 활용하여 실제 도구 연동과 실시간 스트리밍이 가능한 오픈소스 에이전트 워크스페이스 구축 과정을 다룹니다. ReAct 루프를 상태 머신으로 모델링하고, SSE를 통해 원격 MCP 서버의 도구를 동적으로 발견 및 통합하는 방법을 제시합니다.
핵심 포인트
- LangGraph로 ReAct 사이클을 그래프화하여 안정적인 에이전트 워크플로우 구현
- SSE와 FastMCP를 이용해 외부 도구를 동적으로 연결하고 관리 가능
- FastAPI를 활용하여 토큰, 도구 시작 등 이벤트를 실시간 스트리밍 지원
- Human-in-the-loop 승인 게이트를 통해 민감한 작업에 대한 안전성 확보
대부분의 에이전트 튜토리얼은 노트북에서 짧고 선형적인 체인(chain)을 보여줍니다. 저는 실제 도구를 사용하고, 라이브 스트리밍 및 안전 검사를 거치는 에이전트를 프로덕션 환경에서 실행하는 것이 어떤지 보고 싶어서 Agentic AI Workspace를 구축했습니다. 이 워크스페이스는 LangGraph (ReAct 루프) 에이전트와 MCP 도구 서버(FastMCP), FastAPI, 그리고 Next.js 14로 구성되어 있습니다. 현재 라이브이며 오픈 소스로 공개되었습니다.
투명성: 저는 AI의 도움을 받아 이것을 구축했습니다. 아키텍처, 도구 세트 및 안전 게이트를 설계했고, 직접 배포하고 테스트했습니다. 제가 사용한 4가지 패턴(간소화된 발췌)은 다음과 같습니다.
- 상태 머신으로서의 순환 ReAct 루프
수동으로 작성하는while루프 대신, LangGraph는 이 사이클을 그래프로 모델링합니다.add_messages리듀서가 모든 도구 턴(tool turn)을 상태에 추가합니다.
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
# LLM이 도구를 요청하면 실행하고, 그렇지 않으면 그래프를 종료합니다.
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent") # 도구 결과는 모델로 다시 전달됩니다.
- 런타임에서 MCP 도구 발견하기
백엔드는 SSE를 통해 MCP 서버에 연결하고, 해당 서버의 도구를 목록화한 다음, 각 도구를 LangChain 도구로 감싸서(wrap) 사용합니다. 따라서 모든 통합을 하드코딩할 필요가 없습니다.
async with sse_client(url) as (read_stream, write_stream):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
result = await session.list_tools()
for t in result.tools:
# 원격 MCP 도구를 LangChain StructuredTool로 감쌉니다.
lc_tool = StructuredTool.from_function(coroutine=_call, name=f"{server}__{t.name}")
동일한 FastMCP 서버는 Claude Desktop 및 Cursor에도 플러그인할 수 있습니다 (첫 번째 댓글에 설정 포함).
- SSE를 통한 토큰 및 도구 이벤트 스트리밍
아무도 로더(loader) 화면을 10초 동안 응시하는 것을 좋아하지 않기 때문에, FastAPI는 LangGraph 이벤트를 발생하는 즉시 스트리밍합니다.
async for event in agent.astream_events({"messages": history}, version="v2"):
if event["event"] == "on_chat_model_stream":
yield f"data: {json.dumps({'type': 'token', 'content': token})}
"
elif event["event"] == "on_tool_start":
yield f"data: {json.dumps({'type': 'tool_start', 'name': event['name']})}
- Human-in-the-loop 및 no eval() 승인 게이트:
Gmail SMTP를 통한 이메일 전송과 같은 작업은 명시적인 승인이 필요합니다. 에이전트는 UI에 초안 카드(draft card)를 생성하고, 사용자가 '승인 및 전송(Approve & Send)'을 클릭할 때까지 실행이 중단됩니다.
eval() 없는 수학 계산 : 표현식은 Python의 ast 모듈과 사용자 지정 연산자 화이트리스트로 구문 분석되므로 임의 코드가 실행되지 않습니다.
3가지 프로덕션에서 문제가 발생했던 사례
Gemini의 메시지 역할 교대: 제 설정에서는 동일한 역할을 가진 연속적인 메시지가 거부되었습니다. SSE 연결이 끊긴 후 다음 쿼리가 조용히 실패했습니다. 저는 그래프에 도달하기 전에 연속된 동일 역할 메시지를 병합하는 인터셉터(interceptor)를 추가했습니다.
MCP stdio 손상: MCP stdio 서버에서 어떤 것이든 stdout에 쓰면 JSON-RPC 스트림이 깨집니다. 해결책은 모든 로깅을 sys.stderr로 전송하는 것이었습니다.
Render의 무료 티어 512MB RAM: Next.js SSR과 Python 백엔드가 메모리 부족 상태가 되었습니다. 저는 Next.js 번들을 정적으로 내보내고 프론트엔드와 API를 하나의 FastAPI 프로세스에서 제공하는 다단계 Docker 빌드(multi-stage Docker build)로 전환했습니다. ([측정한 유휴 메모리]).
기술 스택: LangGraph · FastMCP · FastAPI · Next.js 14 · MongoDB (TTL memory) · Docker
피드백 환영합니다. 특히 도구 설계에 대한 피드백 부탁드립니다.
LangGraph 또는 MCP를 프로덕션에서 사용할 때 가장 큰 문제점은 무엇인가요?
/u/Entire_Square_2148 님이 제출함 [링크] [댓글]
AI 자동 생성 콘텐츠
본 콘텐츠는 r/SelfHosted (AI filter)의 원문을 AI가 자동으로 요약·번역·분석한 것입니다. 원 저작권은 원저작자에게 있으며, 정확한 내용은 반드시 원문을 확인해 주세요.
원문 바로가기