3,994
🎈 이 글 1부에서는 앱을 지탱하는 Streamlit 기반 메커니즘을, 2부에서는 Agents SDK가 제공하는 Hosted Tools와 MCP에 대해 공부한 내용에 대한 기록입니다.
Streamlit 앱은 사용자가 무언가를 할 때마다(interaction) Python 파일 전체를 처음부터 다시 실행한다.
agent = Agent(...) 같은 코드를 그대로 사용하면 매번 새 Agent 객체가 생성되고 이전 대화 내용이나 상태가 날아간다.
이를 해결하기 위해 st.session_state 를 사용하는데, 이는 재실행 사이에 살아있는 저장소로 딕셔너리처럼 생겼다.
# 딕셔너리처럼 key로 접근
st.session_state["agent"] = Agent(...) # 저장
print(st.session_state["agent"]) # 읽기
그래서 현재 코드가 이런 패턴:
if "agent" not in st.session_state: # 처음 실행할 때만
st.session_state["agent"] = Agent( # Agent를 딱 한 번 생성
name="Life Coach Agent",
...
)
agent = st.session_state["agent"] # 이후 재실행에선 저장된 것 가져오기웹 검색 같은 도구 호출 상태를 화면에 보여주는 코드는 크게 세 위치에서 동작한다.
# 위치 A: paint_history() 안 (과거 메시지 재렌더링)
with st.chat_message("ai"):
st.write("🚀 웹 검색 ...")
# 위치 B: run_agent() 안 (새 메시지 처리 중)
status_container = st.status("🌿", expanded=False)
# 위치 C: run_agent() 안 (스트리밍 이벤트 처리)
status_container.update(label=f"🔍 웹 검색: {query}", state="complete")
update_status(status_container, event.data.type) # "✅ 웹 검색 완료" 등[앱 로드]
└─ paint_history() 실행
└─ 과거 메시지 순서대로 그림
└─ web_search_call 타입이면 "🚀 웹 검색 ..." 표시 ← 위치 A
[유저가 메시지 입력]
└─ run_agent() 실행
└─ st.status("🌿") 생성 ← 위치 B (초기 로딩 표시)
└─ 스트리밍 시작...
└─ tool_called 이벤트 → "🔍 웹 검색: {query}" ← 위치 C
└─ web_search_call.in_progress → "🤖 웹 검색 시작..." ← 위치 C
└─ web_search_call.searching → "⏳ 웹 검색 중..." ← 위치 C
└─ web_search_call.completed → "✅ 웹 검색 완료" ← 위치 CStreamlit의 대부분 함수는 화면에 뭔가를 그리고 끝.
st.write("hello") # 화면에 쓰고 끝, 나중에 바꿀 수 없음
st.empty() # 빈 자리 만들고 끝그러나 st.status()는 나중에 업데이트할 수 있는 컴포넌트를 반환한다. 그렇기 때문에 변수에 저장해 두고 .update()로 같은 컴포넌트를 조작한다.
async def run_agent(message):
with st.chat_message("ai"): # 1. AI 말풍선 열기
status_container = st.status("🌿", expanded=False) # 2. status 컴포넌트
text_placeholder = st.empty() # 3. 텍스트 자리 확보
response = "" # 4. 응답 누적 변수
stream = Runner.run_streamed(agent, message, session=session) # 5. 스트림 시작
async for event in stream.stream_events(): # 6. 이벤트 루프
...st.chat_message("ai") 블록 안에서 만들어지는 것들은 모두 말풍선 안에 들어간다. st.empty()는 빈 자리를 미리 예약하는데, 텍스트가 스트리밍으로 조금씩 올 때 이 자리에 계속 덮어쓰기 위해서다. st.write()를 그냥 쓰면 매번 새 줄이 추가되지만, st.empty()를 쓰면 같은 자리에 업데이트된다.
stream = Runner.run_streamed(agent, message, session=session)
Agent를 실행하되, 결과를 한 번에 받지 않고 이벤트 단위로 흘려보내는 스트림을 반환한다. 실제 실행은 아직 시작되지 않으며, stream_events()를 순회할 때 시작된다.
async for event in stream.stream_events():
if event.type == "run_item_stream_event":
if event.name == "tool_called":
query = event.item.raw_item.action.query
status_container.update(label=f"🔍 웹 검색: {query}", state="running")
if event.type == "raw_response_event":
update_status(status_container, event.data.type)
if event.data.type == "response.output_text.delta":
response += event.data.delta
text_placeholder.write(response)response += event.data.delta가 핵심이다. 토큰이 올 때마다 누적하고, 누적된 전체를 text_placeholder.write(response)로 덮어씀. 그래서 타이핑되는 것처럼 보인다.
1부가 "앱의 뼈대"였다면, 2부는 에이전트에게 능력을 붙이는 단계다. OpenAI Agents SDK는 직접 함수를 만들지 않아도 되는 Hosted Tools를 제공한다.
💡 Hosted Tools란? OpenAI가 미리 정의해 둔 내장 도구. 내 컴퓨터가 아니라 OpenAI 서버에서 구동된다.
WebSearchTool,FileSearchTool,ImageGenerationTool,CodeInterpreterTool등이 여기 속한다.
어떤 도구가 언제 호출될지는 결국 에이전트의 instructions 프롬프트가 결정한다. 도구를 등록하는 것과, 그 도구를 언제 쓰라고 알려주는 것은 별개다.
You are a helpful assistant.
You have access to the following tools:
- Web Search Tool: Use this when the user asks a question that
isn't in your training data. Use this to learn about current events.
- File Search Tool: Use this tool when the user asks a question
about facts related to themselves. Or about specific files.
- Code Interpreter Tool: Use this tool when you need to write and
run code to answer the user's question.사용자 입력(prompt)
→ Runner.run_streamed(agent, message, session)
→ 스트리밍 이벤트
├─ update_status() : 도구 상태 칩 갱신
└─ placeholder.write() : 본문·코드·이미지 실시간 렌더
→ SQLiteSession : 대화 메모리에 영속화
→ 리렌더 시 paint_history()로 복원이 앱을 관통하는 두 축은 ① 이벤트 스트리밍 처리와 ② 메모리(세션) 영속화다.
| Tool | 실행 위치 | 언제 쓰나 | 핵심 설정 | 결과물 |
|---|---|---|---|---|
| WebSearchTool | OpenAI 서버 | 학습 데이터에 없는 최신 정보 / 시사 | 없음 | 텍스트 |
| FileSearchTool | OpenAI 서버 | 사용자/파일 관련 사실 질문 | vector_store_ids | 텍스트 |
| ImageGenerationTool | OpenAI 서버 | 이미지 생성 요청 | tool_config | base64 이미지 |
| CodeInterpreterTool | 샌드박스 | 코드 실행이 필요한 답변 | container | 코드 + 실행 결과 |
| HostedMCPTool / MCPServerStdio | 원격 / 로컬 | 외부 도구·문서 연동 | URL / command | MCP 호출 결과 |
이미지 처리는 별도 Tool이 아니다. 멀티모달은 모델에 내장된 기능이라, 이미지를 base64로 인코딩해 메모리에 직접 추가하면 된다.
file_bytes = file.getvalue()
base64_data = base64.b64encode(file_bytes).decode("utf-8")
data_uri = f"data:{file.type};base64,{base64_data}" # AI 모델용 data URI
await session.add_items([
{
"role": "user",
"content": [
{"type": "input_image", "detail": "auto", "image_url": data_uri}
],
}
])💡 이미지를 하나의 거대한 문자열(data URI) 로 만들어 대화 메모리에 끼워 넣는 것이 핵심이다. 텍스트 메모리와 동일한 통로(
session.add_items)를 쓴다.
💡 왜 — 모델의 학습 데이터에 없는 최신 정보·시사를 답해야 할 때.
WebSearchTool() # 별도 설정 없이 추가만 하면 됨도구 호출 시 발생하는 tool_called 이벤트에서 검색어를 꺼내 상태 칩에 보여준다.
if event.type == "run_item_stream_event":
if event.name == "tool_called":
if hasattr(event.item.raw_item, "action"): # 툴마다 raw_item 클래스가 다름
query = event.item.raw_item.action.query
status_container.update(label=f"🔍 웹 검색: {query}", state="running")⚠️
tool_called이벤트는 모든 종류의 툴 호출에서 발생한다. 그런데raw_item클래스가 툴마다 달라서, 예컨대FileSearchTool에는action속성이 없다. 그래서hasattr(..., "action")체크가 필요하다.
💡 왜 — 사용자 자신이나 특정 파일에 관한 사실을 물을 때. OpenAI **Vector Store(스토리지)**에서 파일을 읽어 답한다.
FileSearchTool(
vector_store_ids=[VECTOR_STORE_ID], # 사용자마다 각각의 store를 만들 수 있음
max_num_results=3, # 상위 3개 파일만 가져옴
)연결까지 4단계로 동작한다.
VECTOR_STORE_ID 확보st.chat_input(accept_file=True, file_type=["txt"]) 로 파일 입력 받기uploaded_file = client.files.create(
file=(file.name, file.getvalue()),
purpose="user_data", # purpose 종류가 많으니 주의
)
client.vector_stores.files.create(
vector_store_id=VECTOR_STORE_ID,
file_id=uploaded_file.id,
)💡 왜 — 텍스트로 이미지를 생성할 때.
tool_config가 필요하다.
ImageGenerationTool(
tool_config={
"type": "image_generation", # 필수 옵션
"quality": "medium", # 생성 이미지 품질
"output_format": "jpeg", # 이미지 확장자
"moderation": "low", # 검열 강도
"partial_images": 1, # 생성 과정을 중간 이미지로 스트리밍
}
)partial_images 설정 덕분에 생성 과정의 중간 이미지가 이벤트로 흘러온다. base64로 오므로 디코딩해서 placeholder에 그린다.
elif event.data.type == "response.image_generation_call.partial_image":
image = base64.b64decode(event.data.partial_image_b64)
image_placeholder.image(image)💡 왜 — 답을 내기 위해 코드를 작성하고 실행해야 할 때. 내 컴퓨터와 격리된 샌드박스에서 실행된다.
CodeInterpreterTool(
tool_config={
"type": "code_interpreter", # 필수
"container": {
"type": "auto",
# "file_ids": [""] # 스토리지 file id를 주면 코드에서 파일 접근 가능
},
}
)코드 작성 과정도 delta로 스트리밍된다. st.code()는 단순 텍스트가 아니라 syntax highlight가 적용된 코드를 그린다.
if event.data.type == "response.code_interpreter_call_code.delta":
code_response += event.data.delta
code_placeholder.code(code_response) # 하이라이트된 코드를 실시간 갱신💡 왜 — Web/File/Image 외의 외부 도구·문서를 에이전트에 연결할 때. MCP(Model Context Protocol)로 표준화된 도구 서버를 붙인다.
| HostedMCPTool | MCPServerStdio | |
|---|---|---|
| 실행 위치 | OpenAI 서버 (원격 MCP 호출) | 로컬 (내 머신에서 서버 구동) |
| 필요한 것 | 서버 URL만 | 실행 명령어 (command • args) |
| 캐싱 | 가능 (session_state) | 불가 — 서버를 띄운 채 agent에 넘겨야 함 |
Hosted MCP — URL만 알면 된다.
HostedMCPTool(
tool_config={
"server_url": "https://mcp.context7.com/mcp",
"type": "mcp",
"server_label": "Context_7",
"server_description": "Use this to get the docs from software projects.",
"require_approval": "never", # 사용 전 승인 단계 해제
}
)Local MCP (Stdio) — uvx로 설치 없이 로컬 서버를 띄운다.
from agents.mcp.server import MCPServerStdio
yfinance_server = MCPServerStdio(
params={"command": "uvx", "args": ["mcp-yahoo-finance"]}
)
# Agent(mcp_servers=[yfinance_server], ...) 처럼 배열로 연결⚠️ 로컬 MCP는 session_state 캐싱을 쓸 수 없다. 서버를 계속 실행해 둔 채로, agent를 만들 때 _실행 중인 서버_를 넘겨줘야 하기 때문이다. (
uvx는npx처럼 설치 없이 바로 실행 /uv·npm은 설치 후 실행)
🧩 async 밖에서 **
await**가 필요할 때 —session.get_items()는 coroutine이라await가 필요하지만,async함수 밖(예: Reset 버튼 핸들러)에서는await를 못 쓴다. 이럴 때asyncio.run()이await역할을 대신한다.pythonreset = st.button("Reset Memory") if reset: asyncio.run(session.clear_session())
🧩 placeholder를 session에 캐싱하는 이유 — 본문·코드·이미지 placeholder를
st.session_state에 저장해 두고, 새 메시지가 들어올 때만 비운다. 이렇게 해야 이전 응답이 남아 있다가 새 입력 시점에 깔끔히 초기화된다.
🧩 메모리는 dict로 접근 —
paint_history()에서 메시지는 객체가 아니라 dict다.message["role"]처럼 키로 접근해야 하며message.role은 동작하지 않는다. user/assistant, 텍스트/이미지/도구호출 타입별로 분기해 렌더한다.