OpenAI MCP Agents SDK 사용하기

OpenAI MCP Agents SDK 사용하기

 

OpenAI MCP SDK 사용하는 방법을 알아봅니다.

아래 사이트에 가면 가이드가 잘 되어있습니다.

https://openai.github.io/openai-agents-python/mcp/

Model context protocol (MCP) - OpenAI Agents SDK

가상환경 구성 후 mcp.json 파일을 하나 만들고 mcp 서버중 하나를 추가합니다.

https://github.com/modelcontextprotocol/servers/tree/e181222154172e8a9eb38ac09f4af777b36d733e/src

servers/src at e181222154172e8a9eb38ac09f4af777b36d733e · modelcontextprotocol/servers

여기에 있는 항목 중 저는 이전 포스팅에 사용했던 image_reader와 filesystem 을 사용했습니다.

mcp.json

{
    "mcpServers":{
        "image_reader": {
            "command": "npx",
            "args": [
                "-y",
                "mcp-image-reader"
            ]
        },
		"filesystem": {
		  "command": "npx",
		  "args": [
			"-y",
			"@modelcontextprotocol/server-filesystem",
			"C:\\Users\\junij\\Downloads\\pic"
		  ]
		}
    }
}

image_reader 는 이미지를 분석해 주고 filesystem 은 파일을 조작해 줍니다.

원격 서버를 사용해 볼꺼라 바로 Client 작업을 통해 대화를 시도해 봅니다.

먼저 필요한 패키지를 설치합니다.

pip install mcp openai-agents streamlit youtube-transcript-api python-dotenv

환경변수에 OPENAI_API_KEY 를 추가합니다.
(API KEY 는 https://platform.openai.com/api-keys 로 들어가 발급받으면 됩니다.)

.env 파일을 만들고 API_KEY 를 입력합니다.

OPENAI_API_KEY=sk-proj****

mcp_client.py 파일을 만들고 아래처럼 코딩합니다.
(아래 코딩에 대한 설명은 주석으로 대신합니다.)

# pip install mcp openai-agents streamlit youtube-transcript-api python-dotenv
import sys
import asyncio
import streamlit as st
import json
from openai.types.responses import ResponseTextDeltaEvent
from agents import Agent, Runner  # OpenAI Agents 라이브러리 활용
from agents.mcp import MCPServerStdio  # MCP(Model-Control-Protocol) 서버 구현
from dotenv import load_dotenv  # 환경 변수 로드를 위한 라이브러리
load_dotenv()  # .env 파일에서 환경 변수 로드

# Windows 운영체제 호환성을 위한 이벤트 루프 정책 설정
if sys.platform == "win32":
    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())

# MCP 서버 설정 함수
async def setup_mcp_servers():
    """MCP 서버를 설정하고 연결하는 비동기 함수"""
    servers = []
    
    # mcp.json 설정 파일에서 서버 구성 정보를 읽어옴
    with open('mcp.json', 'r') as f:
        config = json.load(f)
    
    # 구성된 각 MCP 서버에 대해 반복 처리
    for server_name, server_config in config.get('mcpServers', {}).items():
        # MCPServerStdio 객체 생성 및 설정
        mcp_server = MCPServerStdio(
            params={
                "command": server_config.get("command"),  # 실행할 명령어
                "args": server_config.get("args", [])     # 명령어 인자
            },
            cache_tools_list=True  # 도구 목록 캐싱 활성화
        )
        # 서버 연결
        await mcp_server.connect()
        servers.append(mcp_server)

    return servers


# 에이전트 설정 함수
async def setup_agent():
    """OpenAI 에이전트를 생성하고 설정하는 비동기 함수"""
    # MCP 서버 설정
    mcp_servers = await setup_mcp_servers()
    
    # 에이전트 객체 생성
    agent = Agent(
        name="Assistant",  # 에이전트 이름
        instructions="궁금한 내용에 대해서 답변을 해주는 에이전트야",  # 에이전트 지시사항
        model="gpt-4o-mini",  # 사용할 모델
        mcp_servers=mcp_servers  # 연결할 MCP 서버 목록
    )
    return agent, mcp_servers


# 사용자 메시지 처리 함수
async def process_user_message():
    """사용자 메시지를 처리하고 에이전트 응답을 스트리밍하는 비동기 함수"""
    # 에이전트와 MCP 서버 설정
    agent, mcp_servers = await setup_agent()
    # 세션에서 대화 기록 가져오기
    messages = st.session_state.chat_history

    # 에이전트 실행 및 응답 스트리밍
    result = Runner.run_streamed(agent, input=messages)

    response_text = ""
    placeholder = st.empty()  # 응답을 표시할 빈 컨테이너

    # 이벤트 스트림 처리
    async for event in result.stream_events():
        # LLM 응답 토큰 스트리밍 처리
        if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
            response_text += event.data.delta or ""
            with placeholder.container():
                with st.chat_message("assistant"):
                    st.markdown(response_text)

        # 도구 호출 및 기타 이벤트 처리
        elif event.type == "run_item_stream_event":
            item = event.item

            if item.type == "tool_call_item":
                tool_name = item.raw_item.name
                st.toast(f"🛠 도구 활용: `{tool_name}`")  # 도구 사용 시 토스트 메시지 표시

    # 완료된 응답을 대화 기록에 추가
    st.session_state.chat_history.append({
        "role": "assistant",
        "content": response_text
    })
    
    # MCP 서버 명시적 종료 (Streamlit에서 비동기 처리 오류 방지)
    for server in mcp_servers:
        await server.__aexit__(None, None, None)

# Streamlit UI 메인 함수
def main():
    """Streamlit 애플리케이션 메인 함수"""
    # 페이지 설정
    st.set_page_config(page_title="나만의 MCP TEST AGENT", page_icon="🤖")

    # 대화 기록 초기화
    if "chat_history" not in st.session_state:
        st.session_state.chat_history = []

    # 제목 및 설명 표시
    st.title("🤖 나만의 MCP TEST AGENT")
    st.caption("MCP 도구를 테스트 하세요!")

    # 기존 대화 기록 표시
    for m in st.session_state.chat_history:
        with st.chat_message(m["role"]):
            st.markdown(m["content"])

    # 사용자 입력 처리
    user_input = st.chat_input("대화를 해주세요")
    if user_input:
        # 사용자 메시지를 대화 기록에 추가
        st.session_state.chat_history.append({"role": "user", "content": user_input})
        with st.chat_message("user"):
            st.markdown(user_input)

        # 비동기로 에이전트 응답 처리
        asyncio.run(process_user_message())
        
# 메인 함수 실행
if __name__ == "__main__":
    main()

실행합니다.

특정 파일에 대한 내용을 분석을 요청하면 우측 상단에 사용되는 도구가 표시되고

이미지에 분석이 됩니다. (캡쳐를 못했네요 ㅜㅠ)

이제 로컬 폴더의 이미지를 알아보고 이미지의 내용에 따라 파일 명을 바꾸도록 해봅니다.

원본파일은 의미없는 이름으로 되어있었으나 이미지를 분석해서 파일명이 실제로 바뀐 것을 알 수 있습니다.