[MCP] LLM과 MCP 통합하기 (OpenAI)

https://deahan.tistory.com/476

[MCP] Server, Client 기본 (Python)

기본을 만들어 봤다면 이제 LLM과 MCP를 통합해보는 과정을 알아보자

 

 

터미널을 열고 프로젝트 디렉토리(D:\project\my-mcp-server)에서 아래 명령어를 실행하여 패키지를 설치

uv add openai python-dotenv

- openai: OpenAI API를 사용하기 위한 패키지
- python-dotenv: 환경 변수를 관리하기 위한 패키지, API 키를 안전하게 저장하고 불러오는 데 사용함

 

실행 결과

Resolved 40 packages in 619ms
Prepared 2 packages in 724ms
░░░░░░░░░░░░░░░░░░░░ [0/4] Installing wheels...
...
Installed 4 packages in 306ms
 + distro==1.9.0
 + jiter==0.10.0
 + openai==1.88.0
 + tqdm==4.67.1

---

API 키 설정하기

-API 키를 안전하게 관리하기 위해 프로젝트 디렉토리(D:\project\my-mcp-server)에 .env 파일을 생성하고 API 키를 저장

OPENAI_API_KEY=your_openai_api_key_here

 

- your_openai_api_key_here 부분에 본인 OpenAI API 키를 입력

 

중요!

.env 파일에는 중요한 정보(API 키 등)가 포함되어 있으므로 보안에 각별히 주의
.gitignore 파일에 .env 항목을 추가하여, 해당 파일이 GitHub 등 원격 저장소에 업로드되지 않도록 설정하는 것이 좋음

 

 

LLM에게 질의하는 코드 작성

client_with_openai.py라는 파일을 생성하여 LLM에게 질의하는 코드를 작성

import asyncio
from dotenv import load_dotenv
load_dotenv()

from openai import AsyncOpenAI
client = AsyncOpenAI()

async def main():
    response = await client.responses.create(
        model="gpt-4o",
        input=[
            {
                "role": "user",
                "content": "3 곱하기 7은?"
            }
        ]
    )
    print(response.output_text)

if __name__ == "__main__":
    asyncio.run(main())

 

코드 설명
-load_dotenv(): .env 파일에서 환경 변수를 로드
-client = AsyncOpenAI(): OpenAI의 비동기 클라이언트 객체를 생성
-client.responses.create(...): LLM에게 질의를 보내고 응답을 받음

 

실행 하기

uv run client_with_openai.py

 

실행 결과

3 곱하기 7은 21입니다.

---

사용자 질의 입력받기

앞서 작성한 코드는 고정된 질문을 LLM에 전달해 결과를 출력하는 방식.

이제 사용자가 직접 질문을 입력하면, 해당 질문을 LLM에 전달하여 응답을 받아 출력하는 코드를 작성해 보겠다. 이를 위해 client_with_openai.py 파일을 아래와 같이 수정

from dotenv import load_dotenv
load_dotenv()

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def query_llm(question: str) -> str:
    resp = await client.responses.create(
        model="gpt-4o",
        input=[{"role": "user", "content": question}]
    )
    return resp.output_text

async def main():
    while True:
        question = input("질문을 입력하세요 (exit 입력 시 종료): ")
        if question.strip().lower() == "exit":
            break
        answer = await query_llm(question)
        print(f"\nLLM 답변 ▶ {answer}\n")

if __name__ == "__main__":
    asyncio.run(main())

 

 

실행 하기

uv run client_with_openai.py

 

질의 하기

3 곱하기 7은?
대한민국의 수도는?
exit

 

실행 결과

질문을 입력하세요 (exit 입력 시 종료): 3 곱하기 7은?

LLM 답변 ▶ 3 곱하기 7은 21입니다.

질문을 입력하세요 (exit 입력 시 종료): 대한민국의 수도는?

LLM 답변 ▶ 대한민국의 수도는 서울입니다.

질문을 입력하세요 (exit 입력 시 종료): exit

사용자가 입력한 질문에 대해 LLM이 답변을 출력하는 코드가 완성되었습니다.

 

---

MCP 클라이언트 추가하기 - MCP 서버의 도구 목록 가져오기

이제 MCP 클라이언트를 추가하고, MCP 서버와 통신하여 등록된 도구 목록을 가져와보자. client_with_openai.py 파일을 아래와 같이 수정한다.

from dotenv import load_dotenv
load_dotenv()

import asyncio
from openai import AsyncOpenAI
from fastmcp import Client as MCPClient

llm_client = AsyncOpenAI()              # OpenAI LLM
mcp_client = MCPClient("server.py")     # MCP 클라이언트 객체 생성

# LLM 질의
async def query_llm(question: str) -> str:
    resp = await llm_client.responses.create(
        model="gpt-4o",
        input=[{"role": "user", "content": question}]
    )
    return resp.output_text


async def main():
    # MCP 클라이언트
    async with mcp_client:
        print(f"MCP connected → {mcp_client.is_connected()}") # MCP 서버 연결 상태 출력

        tools = await mcp_client.list_tools() # MCP 서버의 도구 목록 가져오기
        print("MCP 서버 도구 목록")
        for tool in tools:
            print(f"tool: {tool}")

        while True:
            question = input("\n질문을 입력하세요 (exit 입력 시 종료): ")
            if question.strip().lower() == "exit":
                break

            answer = await query_llm(question)
            print(f"\nLLM 답변 ▶ {answer}")

    print(f"\nMCP connected → {mcp_client.is_connected()}")

if __name__ == "__main__":
    asyncio.run(main())

 

코드 설명
- mcp_client = MCPClient("server.py"): server.py에 정의된 MCP 서버와 통신하기 위한 클라이언트를 생성
- async with mcp_client:: 비동기 방식으로 MCP 서버에 연결하고, 사용이 끝나면 자동으로 연결을 해제
- await mcp_client.list_tools(): MCP 서버에 등록된 도구 목록을 가져옴

 

실행 결과

MCP connected → True
MCP 서버 도구 목록
tool: name='multiply' description='Multiplies two numbers together.' inputSchema={'properties': {'a': {'title': 'A', 'type': 'number'}, 'b': {'title': 'B', 'type': 'number'}}, 'required': ['a', 'b'], 'type': 'object'} annotations=None

질문을 입력하세요 (exit 입력 시 종료):

 

이렇게 MCP 서버에 연결하여 서버의 도구 목록을 가져왔다.

---

MCP 도구 목록을 LLM이 이해하는 JSON 형식으로 변환하기

MCP 서버에서 가져온 도구 목록은 그대로 LLM에게 전달할 수 없다. OpenAI Responses API가 이해할 수 있는 JSON 형식으로 변환해줘야 한다. 다음은 그 변환 함수와 적용 예시다.

 

from dotenv import load_dotenv
load_dotenv()

import asyncio
from typing import Any, Dict, List
from openai import AsyncOpenAI
from fastmcp import Client as MCPClient

llm_client = AsyncOpenAI()
mcp_client = MCPClient("server.py")

# MCP -> OpenAI 툴 스키마 변환
def to_openai_schema(tool) -> Dict[str, Any]:
    # 입력 스키마 추출
    raw_schema = (
        getattr(tool, "inputSchema", None)
        or getattr(tool, "input_schema", None)
        or getattr(tool, "parameters", None)
    )

    # 다양한 형태를 dict(JSON-Schema) 로 통일
    if raw_schema is None:
        schema: Dict[str, Any] = {"type": "object", "properties": {}, "additionalProperties": True}

    elif isinstance(raw_schema, dict):
        schema = raw_schema

    elif hasattr(raw_schema, "model_json_schema"):          # Pydantic v2 모델
        schema = raw_schema.model_json_schema()

    elif isinstance(raw_schema, list):                      # list[dict]
        props, required = {}, []
        for p in raw_schema:
            props[p["name"]] = {
                "type": p["type"],
                "description": p.get("description", ""),
            }
            if p.get("required", True):
                required.append(p["name"])
        schema = {"type": "object", "properties": props}
        if required:
            schema["required"] = required

    else:                                                   # 알 수 없는 형식
        schema = {"type": "object", "properties": {}, "additionalProperties": True}

    # 필수 키 보강
    schema.setdefault("type", "object")
    schema.setdefault("properties", {})
    if "required" not in schema:
        schema["required"] = list(schema["properties"].keys())  # 모두 optional 로 두고 싶다면 []

    # OpenAI 툴 JSON 반환
    return {
        "type": "function",
        "name": tool.name,
        "description": getattr(tool, "description", ""),
        "parameters": schema,
    }


async def query_llm(question: str) -> str:
    resp = await llm_client.responses.create(
        model="gpt-4o",
        input=[{"role": "user", "content": question}]
    )
    return resp.output_text


async def main():
    async with mcp_client:
        print(f"MCP connected → {mcp_client.is_connected()}")

        mcp_tools = await mcp_client.list_tools()
        print("MCP 서버 도구 목록")
        for tool in mcp_tools:
            print(f"tool: {tool}")

        # MCP 도구를 OpenAI 툴 스키마로 변환
        for tool in mcp_tools:
            print(f"OpenAI tool schema: {to_openai_schema(tool)}")

        while True:
            question = input("\n질문을 입력하세요 (exit 입력 시 종료): ")
            if question.strip().lower() == "exit":
                break

            answer = await query_llm(question)
            print(f"\nLLM 답변 ▶ {answer}")

    print(f"\nMCP connected → {mcp_client.is_connected()}")


if __name__ == "__main__":
    asyncio.run(main())

 

코드 설명

- to_openai_schema(tool): MCP 도구를 OpenAI Responses API에서 이해할 수 있는 JSON 형식으로 변환하는 함수.

- 이 함수는 MCP 도구 정의를 OpenAI API에서 요구하는 function 형식의 JSON으로 바꿔주는 역할을 한다. 코드를 모두 이해하려고 애쓰기보다는 “MCP → OpenAI 도구 변환 함수” 정도로 가볍게 인식하면 좋다

 

실행 결과

MCP connected → True
MCP 서버 도구 목록
tool: name='multiply' description='Multiplies two numbers together.' inputSchema={'properties': {'a': {'title': 'A', 'type': 'number'}, 'b': {'title': 'B', 'type': 'number'}}, 'required': ['a', 'b'], 'type': 'object'} annotations=None
OpenAI tool schema: {'type': 'function', 'name': 'multiply', 'description': 'Multiplies two numbers together.', 'parameters': {'properties': {'a': {'title': 'A', 'type': 'number'}, 'b': {'title': 'B', 'type': 
'number'}}, 'required': ['a', 'b'], 'type': 'object'}}

질문을 입력하세요 (exit 입력 시 종료):

 

OpenAI tool schema: OpenAI의 LLM 이 이해할 수 있는 JSON 형식으로 도구 목록이 변환됐다.

---

변환된 MCP 도구 목록을 LLM에게 전달하기

이제 query_llm() 함수에 변환된 도구 목록을 함께 전달하여, LLM이 상황에 따라 적절한 MCP 도구를 선택해 호출할 수 있도록 해보자

from dotenv import load_dotenv
load_dotenv()

import asyncio, json
from typing import Any, Dict, List
from openai import AsyncOpenAI
from fastmcp import Client as MCPClient

llm_client = AsyncOpenAI()
mcp_client = MCPClient("server.py")


def to_openai_schema(tool) -> Dict[str, Any]:
    raw_schema = (
        getattr(tool, "inputSchema", None)
        or getattr(tool, "input_schema", None)
        or getattr(tool, "parameters", None)
    )

    if raw_schema is None:
        schema: Dict[str, Any] = {"type": "object", "properties": {}, "additionalProperties": True}

    elif isinstance(raw_schema, dict):
        schema = raw_schema

    elif hasattr(raw_schema, "model_json_schema"):
        schema = raw_schema.model_json_schema()

    elif isinstance(raw_schema, list):
        props, required = {}, []
        for p in raw_schema:
            props[p["name"]] = {
                "type": p["type"],
                "description": p.get("description", ""),
            }
            if p.get("required", True):
                required.append(p["name"])
        schema = {"type": "object", "properties": props}
        if required:
            schema["required"] = required

    else:
        schema = {"type": "object", "properties": {}, "additionalProperties": True}

    schema.setdefault("type", "object")
    schema.setdefault("properties", {})
    if "required" not in schema:
        schema["required"] = list(schema["properties"].keys())

    return {
        "type": "function",
        "name": tool.name,
        "description": getattr(tool, "description", ""),
        "parameters": schema,
    }


async def query_llm(question: str, tool_schemas: List[Dict[str, Any]]) -> str:
    ########## 1차 요청 ##########
    resp = await llm_client.responses.create(
        model="gpt-4o",
        input=[{"role": "user", "content": question}],
        tools=tool_schemas,
    )

    ##### 툴 호출이 없을 때 #####
    tool_calls = [o for o in resp.output if getattr(o, "type", "") == "function_call"]
    if not tool_calls:
        print("툴 호출 없음, 바로 답변 반환")
        return resp.output_text

    ##### 결과를 담을 next_input 에 user 질문 유지 #####
    next_input: List[Any] = [{"role": "user", "content": question}]

    ########## 각 툴 호출 처리 ##########
    for call in tool_calls:
        # MCP 서버 실행 (arguments 는 str일 수도 dict일 수도 있음)
        print(f"call.name: {call.name}, call.id: {call.call_id}")
        args = call.arguments
        if isinstance(args, str):
            args = json.loads(args)
            print(f"args (str): {args}")
        result = await mcp_client.call_tool(call.name, args)

        # 호출 자체를 메시지 배열에 추가
        next_input.append(call)
        # 실행 결과를 function_call_output 형식으로 추가
        next_input.append(
            {
                "type": "function_call_output",
                "call_id": call.call_id,
                "output": str(result),
            }
        )

    ########## 2차 호출 -> 최종 답변 ##########
    final = await llm_client.responses.create(
        model="gpt-4o",
        input=next_input,
    )
    return final.output_text      

async def main():
    async with mcp_client:
        print(f"MCP connected → {mcp_client.is_connected()}")

        mcp_tools = await mcp_client.list_tools()
        tool_schemas = [to_openai_schema(tool) for tool in mcp_tools]

        while True:
            question = input("\n질문을 입력하세요 (exit 입력 시 종료): ")
            if question.strip().lower() == "exit":
                break

            # LLM에 tool 스키마와 함께 질문
            answer = await query_llm(question, tool_schemas)
            print(f"\nLLM 답변 ▶ {answer}")

    print(f"\nMCP connected → {mcp_client.is_connected()}")


if __name__ == "__main__":
    asyncio.run(main())

 

query_llm 함수의 실행 과정
1. 1차 호출: LLM에게 사용자의 질문과 MCP 도구 목록을 전달하여 답변을 받는다
2. 1차 호출의 답변에 툴 호출이 있는지 확인한다.
  - 툴 호출이 없을 때: LLM이 도구를 호출하지 않은 경우, LLM의 답변을 바로 반환 => 답변 완료
  - 툴 호출이 있을 때: 툴을 호출하고 LLM에 계속 질의를 해야 하므로 초기의 사용자 질문을 "next_input"에 유지.
    - 즉, 대화 이력을 관리하여 맥락을 유지.
3. 툴 호출 처리
  - MCP 클라이언트를 통해 MCP 서버의 도구를 호출.
  - 툴 호출 자체를 "next_input"에 추가하여 LLM이 도구 호출을 인식할 수 있도록 함.
  - 도구 호출의 결과를 "function_call_output" 형식으로 "next_input"에 추가하여 LLM이 도구 호출의 결과를 이해할 수 있도록 함
4. 2차 호출: LLM에게 "next_input"을 전달하여 최종 답변을 요청
  - "next_input"에는 사용자 질문, 도구 호출, 도구 호출 결과가 포함된다.

 

코드 설명
- query_llm(question: str, tool_schemas: List[Dict[str, Any]]): LLM에게 질문과 도구 스키마를 전달하여 응답을 받는 함수
- final = await llm_client.responses.create(...): 도구 호출 결과가 포함된 최종 답변을 LLM에게 요청
- await mcp_client.call_tool(call.name, args): LLM이 호출한 도구를 MCP 서버에 요청하여 결과를 받아옴.
- next_input.append(...): LLM의 다음 입력으로 도구 호출과 결과를 추가.
- final.output_text: LLM의 최종 답변을 반환.

 

질의 하기

3 곱가히 7은?
대한민국의 수도는?
exit

 

실행결과

MCP connected → True

질문을 입력하세요 (exit 입력 시 종료): 3 곱하기 7은?
call.name: multiply, call.id: call_Ny3oAhcSzhJHoj3ClZCLkQg7
args (str): {'a': 3, 'b': 7}

LLM 답변 ▶ 3 곱하기 7은 21입니다.

질문을 입력하세요 (exit 입력 시 종료): 대한민국의 수도는?
툴 호출 없음, 바로 답변 반환

LLM 답변 ▶ 대한민국의 수도는 서울입니다.

질문을 입력하세요 (exit 입력 시 종료): exit

 

이제 LLM이 사용자의 질문을 이해하고, 필요에 따라 MCP 서버의 도구를 호출하여 정확한 응답을 생성하는 MCP 호스트를 구축하게 되었다. 코드가 길고 복잡할 수 있지만, 핵심은 LLM <-> MCP 간의 상호작용 흐름을 파악하는 것

 

 

출처 : https://wikidocs.net/book/17905