MCP의 원리와 Claude를 이용한 사용방법을 알아보았지만, 더 근본적으로 MCP Client와 Server가 어떻게 만들지고 사용되는지에 대해 더 깊게이해하고 넘어가야 추후에 응용하는데 도움이 될 것 같아. 유튜브에서 MCP Client, Server를 직접 만들어서 Youtube Agent 데모를 구현한 예제가 있길래, 이걸 보면서 Mcp Client와 Server의 생성방법과 작동 원리에 대해 이해하고 넘어가도록 하겠습니다.
1. MCP 서버 생성
먼저 MCP 서버를 만들어줍니다. Server라고 생각하면 어렵게 느껴질 수 있는데, MCP 서버는 생각보다 단순합니다. 일반 함수를 작성하는 것과 크게 다르지 않습니다.
from mcp.server.fastmcp import FastMCP
# Create an MCP server
mcp = FastMCP("youtube_agent_server")
@mcp.tool()
def get_youtube_transcript(url: str) -> str:
""" 유튜브 영상 URL에 대한 자막을 가져옵니다."""
# 1. 유튜브 URL에서 비디오 ID를 추출합니다.
video_id_match = re.search(r"(?:v=|\/)([0-9A-Za-z_-]{11}).*", url)
if not video_id_match:
raise ValueError("유효하지 않은 YouTube URL이 제공되었습니다")
video_id = video_id_match.group(1)
languages = ["ko", "en"]
# 2. youtube_transcript_api를 사용하여 자막을 가져옵니다.
try:
transcript_list = YouTubeTranscriptApi.get_transcript(video_id, languages=languages)
# 3. 자막 목록의 'text' 부분을 하나의 문자열로 결합합니다.
transcript_text = " ".join([entry["text"] for entry in transcript_list])
return transcript_text
except Exception as e:
raise RuntimeError(f"비디오 ID '{video_id}'에 대한 자막을 찾을 수 없거나 사용할 수 없습니다.{e}")
...
우리가 앞서 MCP Server를 "표준화된 모델 컨텍스트 프로토콜을 활용해 특정 기능을 노출하는 경량 프로그램"이라고 정의하였습니다. 그럼 우리가 MCP Server에서 구현해야할 것은 "특정 기능"이는 것입니다. 이 특정 기능을 함수화 해서 만들어두고 "@mcp.tool()" 데코레이터로 표시해두면 됩니다. 이 데코레이터는 함수자체가 mcp client들이 활용 가능한 형태로 자동으로 변환 시켜주는 역할을 합니다.
핵심 포인트
Tool Calling을 통한 Agent를 구현해 보신 분들은 익숙하겠지만 Agent에게 도구를 여러개 Mapping 시켜줬을때 Agent는 어떤 상황에 어떤 도구를 써야할지 판단을 내리기 힘듭니다. 그래서 도구마다 description을 자세히 적어줄 필요가 있습니다. MCP도 마찬가지입니다.
- 상세한 설명(docstring): LLM이 도구의 목적과 사용법을 정확히 이해하도록 돕습니다.
- 타입 힌트: 인자와 반환값의 타입을 명시하여 LLM이 더 정확하게 도구를 사용할 수 있게 합니다.
if __name__ == "__main__":
print("Starting MCP server...")
mcp.run()
만들고 싶은 도구를 이어서 만들어주고 하단에 mcp.run()으로 mcp server 코드를 마무리합니다.
2. MCP Clinet 연동을 위한 준비
MCP Clinet를 생성하기에 앞서 로컬 MCP 서버를 연동하려면, 서버 실행에 필요한 Python 실행 파일 경로와 MCP 서버 스크립트 경로를 JSON 설정에 입력해야 합니다. 따라서 별도의 json파일을 만들고 하단 Json 파일처럼 작성해줍니다.
{
"mcpServers": {
"mcp-test": {
"command": "/Users/yourname/projects/python_mcp_agent/venv/bin/python",
"args": [
"/Users/yourname/projects/python_mcp_agent/2_mcp_server.py"
]
}
}
}
command에 가상환경의 경로, args에 우리가 생성한 mcp Server의 경로를 작성해줍니다. 경로는 왼쪽 디렉토리 구조에서 우클릭으로 Path를 확인하는게 제일 간편합니다.
3. MCP Client 생성
이제 본격적인 준비가 끝났으니 MCP Clinet를 생성하고 Server와 연동해서 실행해보겠습니다.
from agents.mcp import MCPServerStdio
# MCP 서버 설정
async def setup_mcp_servers():
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():
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
우리가 작성한 json 파일을 가져와서 서버 목록을 확인하고 순회하며 연결합니다.
from agents import Agent, Runner
# 에이전트 설정
async def setup_agent():
# 서버가 이미 존재하는지 확인하고, 없으면 생성
mcp_servers = await setup_mcp_servers()
agent = Agent(
name="Assistant",
instructions="너는 유튜브 컨텐츠 분석을 도와주는 에이전트야",
model="gpt-4o-mini",
mcp_servers=mcp_servers
)
return agent,mcp_servers
여기서 사용한 Agent는 OpenaAI Agents SDK로 Openai에서 정식으로 공개한 패키지입니다. Agent를 간단하게 만들고 MCP Server 연동도 쉽게 할 수 있도록 도와줍니다.
# 메시지 처리
async def process_user_message():
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
})
# 명시적 종료 (streamlit에서 비동기 처리 오류 방지)
for server in mcp_servers:
await server.__aexit__(None, None, None)
streamlit 기본 설정을 마친뒤 위 코드로 사용자의 입력에 대해 "Runner.run_streamed(agent, input=messages)" 실행하고 결과값을 받아와 화면에 보여주면 됩니다.
만들어져있는 MCP Server를 가져다가 사용하는 것도 좋지만, 어떤 원리로 작동하는지를 코드를 뜯어보면서 살펴보니 확실히 이해하는데 더 도움이 되는 것 같습니다.