AI Agent(3/3): MCP(Model Context Protocol) 서버와 MCP 클라이언트(LLM 에이전트)

Agentic AI/AI_AGENT

AI Agent(3/3): MCP(Model Context Protocol) 서버와 MCP 클라이언트(LLM 에이전트)

아톨 2026. 10. 3. 14:57

세번째 코드는 MCP(Model Context Protocol) 서버와 MCP 클라이언트(LLM 에이전트)가 SSE(Server-Sent Events, HTTP 기반) 방식으로 네트워크를 통해 통신하도록 구현된 예제입니다.

이전 Stdio(표준 입출력) 방식과 달리, 서버와 클라이언트가 서로 별도의 프로세스/컴퓨터로 분리될 수 있는 웹 표준 서버 architecture 형태를 띱니다.

두 파일의 구조와 작동 원리를 핵심 위주로 명쾌하게 해설해 드리겠습니다.

1. 시스템 구조 개요

┌────────────────────────────────┐         HTTP GET/POST (SSE)        ┌────────────────────────────────┐
│   첫 번째 파일: MCP Server    │ <================================> │  두 번째 파일: MCP Client      │
│   (WeatherExpert / FastMCP)    │   http://localhost:8000/sse       │  (OpenAI responses API Agent)  │
└────────────────────────────────┘                                   └────────────────────────────────┘
    │  - @mcp.tool() : 실시간 날씨 API                                        │  - session.get_prompt()
    │  - @mcp.resource() : 기상 정책 지침                                      │  - session.read_resource()
    └  - @mcp.prompt() : 기상 캐스터 페르소나                                   └  - session.call_tool()

2. 첫 번째 파일 해설 (MCP Server - WeatherExpert)

FastMCP를 사용하여 Tool(도구), Resource(자원), Prompt(프롬프트)를 제공하는 독립형 HTTP 서버입니다.

import asyncio

import json

import os

import pathlib

# from mcp import ClientSession, StdioServerParameters

from mcp import ClientSession

# from mcp.client.stdio import stdio_client

from mcp.client.sse import sse_client

from dotenv import load_dotenv

from openai import OpenAI

load_dotenv(override=True)

# BASE_DIR = pathlib.Path(__file__).parent

async def run_gpt_mcp_agent():

# server_params = StdioServerParameters(

# command="python",

# args=["test_6_16_.py"],

# cwd=str(BASE_DIR),

# )

# 1. 접속할 MCP 서버의 SSE 엔드포인트 설정

# 서버가 로컬의 8000번 포트에서 구동 중이라고 가정한다.

server_url = "http://localhost:8000/sse"

# 2. sse_client를 사용하여 서버와 연결 통로(read, write)를 연다.

# async with stdio_client(server_params) as (read, write):

async with sse_client(server_url) as (read, write):

async with ClientSession(read, write) as session:

# 프로토콜 핸드셰이크 및 초기화

await session.initialize()

# [프롬프트 주입] 서버에서 기상 캐스터 페르소나를 가져온다.

mcp_prompt = await session.get_prompt("weather_briefing", arguments={"location":"서울특별시"})

prompt_text = mcp_prompt.messages[0].content.text

# [리소스 주입] 서버에서 대응 지침(Policy)을 읽어온다.

mcp_resource = await session.read_resource("mcp://weather/policy")

policy_text = mcp_resource.contents[0].text

# 메시지 구성: 지침과 페르소나를 모델의 시스템 프롬프트에 결합한다.

input_items = [

{"role":"system","content":f"{prompt_text}\n\n참고 정책:\n{policy_text}"},

{"role":"user", "content":"현재 서울 날씨 확인하고, 정책에 맞춰 브리핑해줘."},

]

# [도구 변환] MCP 도구 명세를 OpenAI Function 규격으로 매핑한다.

mcp_tools = await session.list_tools()

gpt_tools =[

{

"type":"function",

"name":t.name,

"description":t.description,

"parameters":t.input_schema,

} for t in mcp_tools.tools

]

# 3. OpenAI 클라이언트 초기화

openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY",""))

model = os.getenv("OPENAI_MODEL","gpt-5.4-mini")

# 4. 첫 번째 호출: 도구 사용 여부 결정

response = openai_client.responses.create(

model=model,

input=input_items,

tools=gpt_tools if gpt_tools else None,

tool_choice="required",

)

# 5. 도구 실행 및 피드백 루프 (The Loop)

input_items.extend(response.output)

tool_calls = [item for item in response.output if item.type=="function_call"]

if tool_calls:

for tool_call in tool_calls:

tool_name = tool_call.name

tool_args = json.loads(tool_call.arguments)

# MCP 서버에 실제 도구 실행 요청 (네트워크 통신 발생)

result = await session.call_tool(tool_name, tool_args)

result_text = result.content[0].text

# 실행 결과를 메시지 기록에 삽입

input_items.append(

{

"type":"function_call_output",

"call_id":tool_call.call_id,

"output":result_text,

}

)

# 6. 두 번째 호출: 결과를 바탕으로 최종 정책 반영 브리핑 생성

final_response = openai_client.responses.create(

model=model,

input=input_items,

)

print(f"최종 에이전트 브리핑: \n{final_response.output_text}")

else:

print(f"최종 에이전트 브리핑: \n{response.output_text}")

if __name__=="__main__":

asyncio.run(run_gpt_mcp_agent())

① 서버 생성 및 실행

mcp = FastMCP("WeatherExpert")

if __name__ == "__main__":
    mcp.run(transport="sse", host="127.0.0.1", port=8000)

transport="sse": Stdio 통신이 아닌, HTTP 기반 SSE 서버로 작동시킵니다.
host="127.0.0.1", port=8000: [http://127.0.0.1:8000/sse](http://127.0.0.1:8000/sse) 엔드포인트를 열어 클라이언트의 접속을 대기합니다.

② @mcp.tool() - 실시간 날씨 조회 기능

@mcp.tool()
def get_city_weather(location: str) -> str:
    ...

역할: LLM이 실행을 요청할 수 있는 실제 백엔드 연동 기능입니다.
동작 순서:
1. Open-Meteo Geocoding API로 도시 이름(location)의 위도(lat)와 경도(lon)를 변환합니다.
2. Open-Meteo Weather Forecast API로 위도/경도의 실시간 기온, 습도, 기상 코드를 조회합니다.
3. parse_weather_code() 함수를 통해 기상 코드를 한국어 상태(맑음, 비, 눈 등)로 파싱합니다.
4. 결과를 JSON 문자열로 변환하여 클라이언트에 반환합니다.

③ @mcp.resource() - 정적 안내 규칙

@mcp.resource("mcp://weather/policy")
def weather_policy() -> str:
    return """
    - 폭염 주의보: 기온 33도 이상이 2일 이상 지속될 때
    - 불쾌지수 높음: 습도 60% 이상일 때
    - 야외 활동 권장: 기온 20~25도 사이의 맑은 날씨
    """

역할: LLM 판단의 기준이 되는 정책/지침 데이터를 읽기 전용으로 전달합니다.
URI 식별자 (mcp://weather/policy): 웹 URL처럼 클라이언트가 서버 내 자원을 지칭할 때 사용하는 전용 주소 규약입니다.

④ @mcp.prompt() - 시스템 페르소나 템플릿

@mcp.prompt()
def weather_briefing(location: str) -> str:
    return f"""
    너는 대한민국 최고의 기상 캐스터다.
    전문적인 용어를 섞어가며 {location}의 날씨를 시민들에게 친절하게 설명해라.
    만약 리소스에 정의된 대응 지침에 해당한다면 생활 정보도 함께 제공해라.
    """

역할: 프롬프트 엔지니어링 템플릿을 서버가 관리하여, 클라이언트가 이를 동적으로 불러와 대화 맥락으로 채택할 수 있게 해줍니다.

3. 두 번째 파일 해설 (MCP Client / Agent)

HTTP/SSE를 통해 MCP 서버에 접속하고, OpenAI 최신 responses.create API를 활용하여 자율적 에이전트 루프(Tool Call)를 실행합니다.

from fastmcp import FastMCP

import requests

import json

# 1. MCP 서버 인스턴스 생성

mcp = FastMCP("WeatherExpert")

# 4. Tool (도구) 정의: LLM이 실행할 수 있는 실체 기능(함수) - LLM이 필요할 때 직접 인자를 전달하여 실행하는 실제 계산/조회/동작 함수

@mcp.tool()

def get_city_weather(location:str)->str:

print(f"[시스템] 현재 {location}의 날씨 데이터 조회중")

try:

geo_url=f"https://geocoding-api.open-meteo.com/v1/search?name={location}&count=1&language=ko&format=json"

geo_res= requests.get(geo_url, timeout=5).json()

print("="*100)

print(geo_res)

print("="*100)

if not geo_res.get("results"):

return json.dumps({"error":f"'{location}의 위치정보를 찾을 수 없습니다.."})

lat = geo_res["results"][0]["latitude"]

lon = geo_res["results"][0]["longitude"]

location_name = geo_res["results"][0].get("name",location)

print("="*100)

print("lat, lon, location_name: ", lat, lon, location_name)

print("="*100)

weather_url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}&current=temperature_2m,relative_humidity_2m,weather_code&timezone=auto"

w_res = requests.get(weather_url, timeout=5).json()

# print(w_res)

current = w_res.get("current",{})

temp = current.get("temperature_2m")

humidity = current.get("relative_humidity_2m")

code = current.get("weather_code")

# if unit=="fahrenheit" and temp is not None:

# temp = round((temp*9/5)+32, 1)

# temp_unit_str = "°F"

# else:

# temp_unit_str = "°C"

condition = parse_weather_code(code)

result = {

"location":location_name,

"condition":condition,

# "temperature":f"{temp}{temp_unit_str}",

"temperature":f"{temp}도",

"humidity":f"{humidity}%"

}

print("="*100)

print(result)

print("="*100)

return json.dumps(result, ensure_ascii=False)

except Exception as e:

return json.dumps({"error":f"날씨 정보를 가져오는중 에러발생: {str(e)}"}, ensure_ascii=False)

def parse_weather_code(code):

if code == 0:

return "맑음"

elif code in [1,2,3]:

return "구름 조금 / 흐림"

elif code in [45, 48]:

return "안개"

elif code in [51,53,55,61,63,65,80,81,82]:

return "비"

elif code in [71,73,75,77,85,86]:

return "눈"

elif code in [95,96,99]:

return "뇌우"

return "정보없음"

# 2. Resource (리소스) 정의: 정적/동적 정보 제공 - 서버가 가진 데이터, 지침, 파일 내용 등을 읽기 전용으로 전달

@mcp.resource("mcp://weather/policy")

def weather_policy()->str:

return """

- 폭염 주의보: 기온 33도 이상이 2일 이상 지속될 때

- 불쾌지수 높음: 습도 60% 이상일 때

- 야외 활동 권장: 기온 20~25도 사이의 맑은 날씨

"""

# "mcp://weather/policy"의 의미: "MCP 클라이언트(호스트)가 MCP 서버 내의 특정 리소스(Resource)에 접근할 때 사용하는 고유 식별 주소(URI)"입니다.

# 우리가 웹 브라우저에서 인터넷 페이지를 찾을 때 [https://www.google.com](https://www.google.com)이라는 URL 주소를 사용하는 것처럼,

# MCP 규약에서는 서버가 제공하는 특정 자원(파일, 데이터베이스, 정책 문서 등)을 구분하기 위해 mcp:// 시작하는 URI 식별자를 사용합니다.

# mcp://[카테고리 또는 서비스명]/[자원 이름]

# │ │ │

# │ │ └─> policy (정책 문서 자원)

# │ └─> weather (날씨 관련 데이터 그룹)

# └─> MCP 커스텀 프로토콜 스키마

# 서버에 @mcp.resource("mcp://weather/policy")라고 데코레이터를 붙여두면, 클라이언트가 session.read_resource("mcp://weather/policy")로

# 요청을 보낼 때 이 함수가 실행되어 해당 정책 텍스트를 돌려주게 됩니다.

# 3. Prompt (프롬프트 템플릿) 정의: AI 지침 및 템플릿 제공 - LLM에게 부여할 역할, 대화 템플릿 구조를 전달

@mcp.prompt()

def weather_briefing(location:str)->str:

return f"""

너는 대한민국 최고의 기상 캐스터다.

전문적인 용어를 섞어가며 {location}의 날씨를 시민들에게 친절하게 설명해라.

만약 리소스에 정의된 대응 지침에 해당한다면 생활 정보도 함께 제공해라.

"""

if __name__ =="__main__":

mcp.run(transport="sse", host="127.0.0.1", port=8000)

# mcp.run() # 기본값: transport="stdio"

① SSE 네트워크 클라이언트 접속

server_url = "http://localhost:8000/sse"

async with sse_client(server_url) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()

sse_client(server_url): HTTP SSE 프로토콜을 이용해 8000번 포트의 서버와 양방향 통신 채널을 생성합니다.
session.initialize(): 클라이언트와 서버 간 프로토콜 버전을 맞추고 상호 기능을 확인하는 핸드셰이크 과정을 수행합니다.

② MCP 요소(Prompt, Resource, Tool) 동적 수집

# 1. 프롬프트 가져오기
mcp_prompt = await session.get_prompt("weather_briefing", arguments={"location": "서울특별시"})
prompt_text = mcp_prompt.messages[0].content.text

# 2. 리소스 가져오기
mcp_resource = await session.read_resource("mcp://weather/policy")
policy_text = mcp_resource.contents[0].text

# 3. 도구 목록 가져오기 & OpenAI 규격 매핑
mcp_tools = await session.list_tools()
gpt_tools = [
    {
        "type": "function",
        "name": t.name,
        "description": t.description,
        "parameters": t.input_schema,
    } for t in mcp_tools.tools
]

MCP 서버에 저장된 기상 캐스터 페르소나, 기상 정책, 도구 스키마를 네트워크를 거쳐 차례로 받아옵니다.

③ 메시지 구성 & 첫 번째 OpenAI 호출

input_items = [
    {"role": "system", "content": f"{prompt_text}\n\n참고 정책:\n{policy_text}"},
    {"role": "user", "content": "현재 서울 날씨 확인하고, 정책에 맞춰 브리핑해줘."},
]

response = openai_client.responses.create(
    model=model,
    input=input_items,
    tools=gpt_tools if gpt_tools else None,
    tool_choice="required",
)

OpenAI responses API 규격에 맞춰 시스템 프롬프트와 사용자 요청을 하나로 결합합니다.
tool_choice="required" 옵션을 부여하여 모델이 기상 조회를 위해 도구를 반드시 사용할 수 있도록 유도합니다.

④ 도구 실행 및 피드백 루프 (The Loop)

input_items.extend(response.output)            
tool_calls = [item for item in response.output if item.type == "function_call"]

if tool_calls:
    for tool_call in tool_calls:
        tool_name = tool_call.name
        tool_args = json.loads(tool_call.arguments)
        
        # 1. MCP 서버로 네트워크 도구 호출 요청
        result = await session.call_tool(tool_name, tool_args)
        result_text = result.content[0].text
        
        # 2. 실행 결과를 대화 이력에 추가
        input_items.append(
            {
                "type": "function_call_output",
                "call_id": tool_call.call_id,
                "output": result_text,
            }
        )
    
    # 3. 최종 답변 생성 (두 번째 OpenAI 호출)
    final_response = openai_client.responses.create(
        model=model,
        input=input_items,
    )
    print(f"최종 에이전트 브리핑: \n{final_response.output_text}")

원격 도구 호출: LLM이 날씨 함수 호출을 요구하면 session.call_tool을 통해 SSE 네트워크로 MCP 서버의 get_city_weather 함수를 원격 실행시킵니다.
최종 종합 브리핑: 함수 실행으로 돌려받은 날씨 데이터(기온, 습도 등)와 정책 규정(폭염, 습도 조건 등)을 종합하여 기상 캐스터 톤의 완결된 브리핑을 생성합니다.

4. Stdio 방식 vs SSE 방식의 핵심 비교

구분	Stdio 방식	SSE (HTTP) 방식
연결 매체	OS 표준 입출력 (파이프라인)	HTTP / SSE 네트워크 포트
구동 형태	클라이언트가 서버 프로세스를 직접 띄움	서버가 독립된 서비스로 항시 대기
배치 구조	로컬 단일 컴퓨터	클라우드, Docker, 분산 네트워크
적용 사례	Claude Desktop, 로컬 개발 환경	마이크로서비스, 멀티 에이전트 플랫폼

저작자표시 비영리 변경금지 (새창열림)