LLM MCP 동작 원리

좋은 질문이에요! “MCP(Model Context Protocol)”는 LLM 앱이 바깥 세계의 데이터와 도구에 안전하게 붙는 표준 대화 규약이에요. 아래처럼 “누가 누구와, 어떤 순서로, 무엇을 주고받는지”를 단계별로 정리해 드릴게요.

한 줄 요약

LLM 앱(호스트)이 클라이언트 를 통해 하나 이상의 MCP 서버 와 상태 유지 연결을 맺고, JSON-RPC 메시지로 리소스(Resources) ·프롬프트(Prompts) ·툴(Tools)을 발견→호출→결과를 받는 흐름입니다. Model Context Protocol+1

구성요소(역할)

  1. Host(호스트) : 채팅앱/IDE처럼 LLM이 뛰는 “컨테이너”. 여러 MCP 클라이언트 를 만들고 권한·보안을 관리합니다. 전체 대화 맥락도 이쪽이 쥡니다. Model Context Protocol

  2. Client(클라이언트) : 특정 MCP 서버 와 1:1 세션을 맺는 커넥터. 기능(캡퍼빌리티) 교섭과 메시지 라우팅을 담당합니다. Model Context Protocol

  3. Server(서버) : 리소스/프롬프트/툴을 “제공”하는 쪽. 로컬 프로세스일 수도, 원격 서비스일 수도 있어요. Model Context Protocol

통신 방식(전송)

  • 메시지 포맷은 JSON-RPC 2.0. 전송은 주로 stdio(서브프로세스 표준입출력) 또는 Streamable HTTP(POST/GET+SSE 옵션) 두 가지가 표준입니다. Model Context Protocol

세션 수명주기(핵심 단계)

  1. initialize 요청 : 클라이언트가 먼저 “지원 프로토콜 버전+클라이언트 기능”을 보냅니다.

  2. initialize 응답 : 서버가 “서버 기능(리소스/프롬프트/툴 등)”과 서버 정보, 선택적 지시문을 돌려줍니다.

  3. initialized 알림 : 클라이언트가 준비 완료를 통지하고 이후 정상 운용 단계로 진입합니다.

  4. 운용(Operation) : 합의된 기능만 사용합니다(예: tools/list, tools/call 등).

  5. 종료 : 전송 방식에 맞게 그레이스풀하게 끊습니다. Model Context Protocol

서버가 제공하는 3가지 “기본 프리미티브”

  • Resources(리소스) : 파일/문서/DB 스키마 같은 참조 데이터. resources/list로 목록을, resources/read로 내용을 가져옵니다. URI(file://, https://, git:// 등)로 식별되고 필요하면 구독/변경 알림도 됩니다. Model Context Protocol+1

  • Prompts(프롬프트) : 사용자가 골라 쓰는 템플릿/워크플로. prompts/list로 발견하고 prompts/get으로 인자 넣어 받아옵니다. Model Context Protocol

  • Tools(툴) : 모델이 “호출”하는 함수. JSON Schema로 입력(필수/형)을 선언하고, tools/list로 발견→tools/call로 실행→텍스트/이미지/구조화 JSON(옵션) 등으로 결과를 반환합니다. Model Context Protocol

실제 호출 흐름, 단계별 예시

  1. 사용자가 “레포 이슈 찾아줘”라고 말함 → 호스트 가 현재 대화·권한을 바탕으로 적절한 MCP 클라이언트 를 선택. Model Context Protocol

  2. 클라이언트↔서버가 initialize 로 버전·기능을 교섭(예: 서버는 tools/resources 지원, 클라이언트는 sampling 지원). Model Context Protocol

  3. 클라이언트가 tools/list로 사용 가능한 툴을 조회(예: search_issues). Model Context Protocol

  4. (필요 시) resources/list/resources/read로 현재 프로젝트 파일·설정 등 맥락을 취득. Model Context Protocol

  5. LLM 이 툴 설명과 입력 스키마를 이해하고 tools/call에 맞는 인자를 작성. 호스트 UI가 사용자에게 실행/권한을 확인(“사람-검토” 권장). Model Context Protocol

  6. 서버가 외부 API/DB를 호출하고 결과를 JSON-RPC 응답으로 반환(필요하면 리소스 링크나 구조화 JSON 포함). Model Context Protocol

  7. 호스트는 결과를 대화 맥락에 주입(서버는 전체 대화 기록을 직접 읽지 못하도록 분리). Model Context Protocol

  8. LLM이 최종 답변을 생성하고, 추가 액션이 필요하면 위 사이클을 반복.

보안·신뢰 원칙(요점)

  • 명시적 사용자 동의 : 데이터 접근·툴 실행 전 사용자 승인이 필요합니다.

  • 최소 공개 : 서버는 필요한 맥락만 보고, 전체 대화/다른 서버의 상태는 볼 수 없게 설계됩니다.

  • 툴은 잠재적 코드 실행 이므로 UI 경고·승인·로깅·취소·타임아웃 등을 권장합니다. Model Context Protocol+1

  • (HTTP 전송 시) Origin 검증/로컬 바인딩/인증 등 네트워크 보안 수칙을 지켜야 합니다. Model Context Protocol

개발자가 기억하면 좋은 포인트

  • 발견(Discovery) : list 계열 API로 기능을 찾는다 → 호출(Call) : schema에 맞춰 실행한다 → 콘텐츠 주입(Contexting) : 결과를 대화에 넣는다(필요시 리소스 링크/구독). Model Context Protocol+1

  • 버전·기능 교섭 이 항상 먼저이고, 합의된 것만 쓴다. Model Context Protocol

  • 구현은 stdio 가 가장 단순하고, Streamable HTTP 는 다중 연결·SSE 스트리밍·세션 ID로 확장 가능. Model Context Protocol

원하시면, 간단한 “Hello Tool” 서버/클라이언트 템플릿(예: Node/Python)으로 initialize → tools/list → tools/call까지 돌아가는 최소 예제를 바로 만들어 드릴게요.

나에게 GPT-OSS 를 자체적으로 구동하는 서버가 있어. 여기에 MCP를 사용하도록 하려면?

좋아요. “자체 GPT-OSS 서버”를 이미 돌리고 계시다면, 거기에 MCP(모델 컨텍스트 프로토콜)를 붙이는 방법은 크게 두 가지 시나리오가 있어요. 아래를 단계별 로 따라가면 됩니다.

시나리오 A) 내 GPT-OSS 앞에 ‘MCP 호스트(브리지)’를 둔다

가장 일반적인 방식. “호스트(Host)=LLM 런타임 + MCP 클라이언트”가 MCP 서버들의 툴·리소스·프롬프트 를 발견→호출하고, 그 결과를 GPT-OSS 대화에 주입합니다. (MCP 클라이언트/서버 SDK와 전송 방식은 공식 문서·SDK에 있어요.) Model Context Protocol+3

0) 사전 준비 (GPT-OSS 서비스를 OpenAI 호환 API로)

  • vLLM 등으로 GPT-OSS를 OpenAI Chat Completions 호환 으로 띄워두면 편합니다. 예: 
    uv pip install --pre vllm==0.10.1+gptoss --extra-index-url https://wheels.vllm.ai/gpt-oss/
vllm serve openai/gpt-oss-20b

(레포의 가이드를 따르면 자동 다운로드/서빙이 됩니다.) GitHub

  • GPT-OSS는 Harmony 응답 형식 을 전제로 학습됐습니다. Harmony로 메시지를 꾸미고(예: openai-harmony 라이브러리), 툴 정의도 시스템 메시지에 넣어야 도구 호출이 올바르게 동작해요. GitHub

1) 어떤 MCP 전송(transport)을 쓸지 결정

  • 동일 머신/로컬 : stdio(서브프로세스로 서버 실행) → 간단/보안 유리

  • 원격/확장 : Streamable HTTP(POST/GET+선택적 SSE) → 다중 클라이언트/세션/재연결 지원

  • HTTP 쓸 땐 Origin 검증, localhost 바인딩, 인증 을 반드시 설정하세요. (DNS rebinding 등 방지) Model Context Protocol

2) 호스트(브리지) 구성 — 최소 루프

핵심은 아래 루프예요.

  1. MCP 클라이언트 세션 을 열고 tools/list로 사용 가능한 MCP 툴 스키마를 가져옵니다.

  2. 그 스키마를 Harmony 포맷의 “tools 정의”로 시스템 메시지에 주입하고, 사용자 프롬프트와 함께 GPT-OSS 에 보냅니다.

  3. 모델이 tool_use 를 생성하면 → 해당 호출을 tools/call로 MCP 서버에 실행 → 결과를 tool_result 로 다시 모델에 전달 → 모델의 최종 답변까지 반복.
    TypeScript/Python 양쪽 SDK가 이 흐름을 지원합니다. Model Context Protocol+1

참고: TypeScript SDK README에는 HTTP/stdio 서버 예제와 클라이언트 작성 섹션이, Python SDK에는 FastMCP 및 Streamable HTTP를 기존 ASGI 앱에 마운트 하는 예제가 잘 정리돼 있어요(브라우저 클라이언트용 CORS 노출 헤더 포함). GitHub+1

3) 샘플 아키텍처(요지)

  • mcp-bridge(호스트):

    • mcp Python/TS SDK로 MCP ClientSession 생성 → list_tools → 툴 스키마를 Harmony 시스템 메시지에 삽입

    • GPT-OSS(예: vLLM OpenAI 엔드포인트)에 ChatCompletions + Harmony 로 질의

    • 응답 내 tool_use → session.call_tool()로 실행 → 결과를 Harmony tool_result로 다시 모델에게 전달

  • 보안 : MCP HTTP 사용 시 Mcp-Session-Id 헤더·세션 종료, CORS 헤더(특히 Mcp-Session-Id 노출) 설정 권장. GitHub

팁: OpenAI의 Apps/Agents SDK 도 MCP 서버 연동 개념을 문서화해 두었고, OpenAI호환(Chat Completions) 엔드포인트 면 쉽게 맞물립니다(호스트를 직접 만들 때 개념 참조). OpenAI 개발자

시나리오 B) GPT-OSS가 MCP 서버 도 겸하도록 확장

특수한 경우: “GPT-OSS가 가진 자체 도구(브라우저 등)를 MCP 서버 형태로 노출”하여 외부 호스트가 붙게 하는 방식.

  • GPT-OSS 레포에는 도구(tool) 개념(예: 브라우저 search/open/find)과 Harmony에 툴 정의를 넣는 예제가 포함돼 있어, 이를 MCP tools/list/tools/call로 매핑해 MCP 서버 를 만들 수도 있습니다. (프로덕션용 브라우저 백엔드는 직접 교체 권장) GitHub

  • 서버 구현은 TypeScript/Python MCP 서버 SDK 의 템플릿을 사용하면 빠릅니다. (Echo/SQLite 등 예제 포함) GitHub

바로 써먹는 체크리스트 (Step-by-Step)

  1. GPT-OSS 엔드포인트 준비

    • vLLM로 OpenAI 호환 API 띄우기 → Harmony 적용(예: openai-harmony) GitHub

  2. MCP 호스트(브리지) 스캐폴딩

    • (Python) pip/uv add mcp 또는 (Node) npm i @modelcontextprotocol/sdk

    • MCP 클라이언트 로 initialize → tools/list까지 호출되는지 확인 GitHub+1

  3. 툴 스키마 ↔ Harmony 변환기 작성

    • MCP tools/list 결과(JSON Schema)를 Harmony tools 정의 로 시스템 메시지에 삽입

    • 모델 응답의 tool_use를 파싱해 tools/call로 라우팅, 결과를 tool_result로 재주입 GitHub+1

  4. 전송 선택과 보안

    • 로컬은 stdio , 분산/원격은 Streamable HTTP

    • HTTP 시 Origin 검증·로컬 바인딩(개발)·인증 ·CORS(Mcp-Session-Id 노출) 설정 Model Context Protocol+1

  5. 테스트 도구 서버 붙이기

    • 예: GitHub 서버(npx @modelcontextprotocol/server-github), 파일시스템/SQLite/포스트그레스 서버 등

    • 커뮤니티 서버 큐레이션(awesome list)도 참고하세요. GitHub+2

  6. 운영 가드레일

    • 툴 실행 시간/빈도 제한, 사용자 확인(UI 승인), 로깅·감사, 타임아웃/취소·리트라이, 비밀키 주입(ENV)

“가장 빠른” 실습 루트(요약)

  • (1) vLLM로 GPT-OSS 띄우기 → vllm serve openai/gpt-oss-20b GitHub

  • (2) TypeScript SDK Quick Start 서버 하나 켜서 MCP 동작 점검(HTTP/stdio 아무거나) GitHub

  • (3) 간이 브리지 :

    • list_tools → Harmony 시스템 프롬프트에 삽입 → ChatCompletions 호출

    • 응답의 tool_use를 tools/call로 실행하고 tool_result로 되돌려 주기 Model Context Protocol+1

  • (4) 보안/세션 헤더 설정(HTTP라면 Mcp-Session-Id/Origin/CORS) Model Context Protocol+1

함께 쓰면 좋은 레퍼런스

  • MCP 전송 스펙(최근판) : stdio & Streamable HTTP, 보안 주의사항, 세션/재시도 흐름 Model Context Protocol

  • MCP 클라이언트 만들기 튜토리얼(언어별) Model Context Protocol

  • TS/Python 공식 SDK(서버·클라이언트, 예제 다수) GitHub+1

  • GPT-OSS 레포 : vLLM 서빙, Harmony/툴 사용법, 브라우저 툴 예시 GitHub+1

  • 커뮤니티 서버 모음 : GitHub/파일/DB/클라우드 등 MCP 서버 카탈로그 GitHub

원하시면, Python/Node 중 선호하시는 언어로 최소 브리지 코드(≈100~150줄)까지 바로 작성해 드릴게요. 모델 호출은 현재 쓰시는 GPT-OSS 엔드포인트(OpenAI 호환)로 그대로 보내고, MCP 툴 호출 루프를 얹어드립니다.

시나리오 A를 보자. 그럼 결국 for문 돌리라는 건가?

짧은 답: 그냥 “for문 한 번”이 아니라 “툴-호출 루프(reactive loop)”예요.
모델이 tool_use(=function call)을 내보낼 때마다 해당 MCP 툴을 실행하고, 그 결과를 다시 모델에 먹여서 다음 응답을 받는 걸 최대 N회 반복하다가, 더 이상 툴 호출이 없으면 종료합니다.

아래를 단계별로 보면 감이 딱 와요.

Step-by-step: 호스트(브리지) 제어 흐름

  1. MCP 서버와 세션 열기 → tools/list로 툴 스키마 수집

  2. 툴 스키마를 모델에 주입

    • (예: GPT-OSS는 Harmony 스타일이면 그 규격대로, 일반 OpenAI 호환이면 tools/function 형식으로)

  3. 대화 시작 : [system(tool정의), user(사용자요청)]로 1차 호출

  4. 응답 검사

    • 응답에 tool_use(=tool/function call)가 없다 → 최종 답변 으로 종료

    • 있으면 → 각 호출을 MCP tools/call로 실행

  5. tool_result 주입: 각 호출 결과를 모델에 다시 전달 → 3~4를 반복

  6. 종료 조건 :

    • 더 이상 tool_use 없음

    • 안전장치(최대 홉 수, 시간 초과, 토큰 한도, 사용자 취소)

즉, “모델 →(요청)→ MCP 실행 →(결과)→ 모델”이 한 턴을 이루고, 이걸 while 루프 로 굴립니다.
필요하면 한 턴 안의 여러 tool_use를 병렬 로 실행할 수도 있어요.

최소 루프(의사코드)

언어/라이브러리 상관없이 패턴 만 잡아두면 됩니다.

# pseudo: OpenAI 호환 ChatCompletions + MCP client

mcp = connect_mcp_server(...)                # 1) MCP 세션
tools = mcp.list_tools()                     # 2) 툴 스키마

messages = [
  system_msg_from(tools),                    # 2) 모델에 툴 정의 주입 (Harmony/Tools 등)
  {"role":"user","content": user_input}
]

MAX_HOPS = 8
for hop in range(MAX_HOPS):                  # ← 핵심 루프 (for든 while든 상관 없음)
    resp = chat_completions(model, messages, tools=to_openai_tools(tools))
    msg = resp["choices"][0]["message"]

    tool_calls = msg.get("tool_calls", [])
    if not tool_calls:
        # 6) 종료: 최종 답변
        print(msg["content"])
        break

    # 4) MCP 툴 실행 (필요 시 병렬)
    results = []
    for call in tool_calls:
        name = call["function"]["name"]
        args = json.loads(call["function"]["arguments"])
        out = mcp.call_tool(name, args)     # 4) MCP tools/call
        results.append({
          "role": "tool",
          "tool_call_id": call["id"],
          "content": json.dumps(out)
        })

    # 5) 모델에 결과 재주입
    messages.extend([msg, *results])

포인트

  • id 매칭 : tool_call_id ↔ MCP 실행 결과를 1:1로 매칭해야 모델이 문맥을 이어받습니다.

  • 병렬 실행 : 동일 턴의 여러 tool_use는 Promise.all/asyncio.gather로 병렬 처리 가능.

  • 안전장치 : MAX_HOPS, timeout, rate limit, 사용자 확인(UI 승인) 넣기.

  • 스트리밍 : 스트리밍 응답이면, 부분 토큰 중간에 tool_use 신호가 나오기도 하니 파서/이벤트로 처리.

TypeScript 예시(요지)

ts

import OpenAI from "openai";             // GPT-OSS의 OpenAI 호환 엔드포인트로 바꿔주세요
import { connect } from "mcp-client";    // 가상의 MCP 클라이언트 API라고 가정

const client = new OpenAI({
  baseURL: process.env.OSS_BASE_URL,     // 예: http://localhost:8000/v1
  apiKey: process.env.OSS_API_KEY || "dummy"
});

const mcp = await connect({ transport: "stdio" /* 또는 http */ });
const tools = await mcp.listTools();

const messages: any[] = [
  systemMsgFrom(tools),                   // Harmony 또는 tools 정의 삽입
  { role: "user", content: process.argv.slice(2).join(" ") }
];

for (let hop = 0; hop < 8; hop++) {
  const res = await client.chat.completions.create({
    model: "gpt-oss-20b",
    messages,
    tools: toOpenAITools(tools)          // OpenAI function-calling 규격으로 변환
  });

  const msg = res.choices[0].message;
  const calls = msg.tool_calls || [];
  if (!calls.length) {
    console.log(msg.content);
    break;
  }

  const results = await Promise.all(calls.map(async (c) => {
    const args = JSON.parse(c.function.arguments || "{}");
    const out = await mcp.callTool(c.function.name, args);
    return { role: "tool", tool_call_id: c.id, content: JSON.stringify(out) };
  }));

  messages.push(msg, ...results);
}

그래서, “for문 돌리면 끝?”의 정확한 답

  • 개념적으로는 맞아요 : “반복적으로 → 모델 호출 → MCP 툴 실행 → 결과 재주입” 루프를 돕니다.

  • 하지만 단순한 for문 이상 입니다.

    • (1) id 매칭/결과 주입 규약

    • (2) 병렬 실행/에러·타임아웃/재시도

    • (3) 최대 홉·토큰/비용 가드

    • (4) 사용자 승인/로깅/권한 스코프

    • (5) (HTTP 전송 시) 세션·CORS·Origin 검증
      같은 운영 요소가 들어가야 “프로덕션스러운” 루프가 됩니다.

원하시면, 지금 쓰시는 GPT-OSS 엔드포인트 URL 기준 으로 돌아가는 최소 브리지 코드를(파이썬/TS 중 선호로) 바로 짜서 드릴게요—툴 정의 변환부터 tool_result 주입, 병렬 호출/타임아웃까지 넣은 버전으로요.

그니까 그냥 while문 돌리는거잖아

맞아요—본질은 while 루프 예요.
“모델 → 툴 호출 신호 확인 → MCP tools/call 실행 → 결과 재주입”을 툴 호출이 사라질 때까지(또는 최대 홉 수까지) 반복합니다.

동작을 딱 정리하면

  1. tools/list로 MCP 툴 스키마 수집 → 시스템/툴 정의로 모델에 주입

  2. 모델 호출 → 응답에 tool_use(=function call)가 있는지 확인

  3. 있으면 그만큼 MCPtools/call 실행(동턴 다건은 병렬 가능)

  4. 각 결과를 tool_call_id에 1:1 매칭 해 tool 결과 메시지 로 다시 모델에 넣기

  5. 2–4를 while 로 반복 (종료: 더 이상 tool_use 없음, MAX_HOPS, 타임아웃 등)

최소 예(파이썬풍)

messages = [system_msg_from(mcp.list_tools()), {"role":"user","content": user_input}]
MAX_HOPS = 8
hop = 0

while hop < MAX_HOPS:
    resp = chat_completions(model, messages, tools=to_openai_tools(mcp.list_tools()))
    msg  = resp["choices"][0]["message"]
    calls = msg.get("tool_calls", [])

    if not calls:                      # 종료 조건
        print(msg.get("content", ""))
        break

    tool_results = []
    for c in calls:                    # 동턴 다건이면 병렬 처리 권장
        name = c["function"]["name"]
        args = json.loads(c["function"]["arguments"] or "{}")
        out  = mcp.call_tool(name, args)     # MCP tools/call
        tool_results.append({
            "role": "tool",
            "tool_call_id": c["id"],         # ← 반드시 매칭!
            "content": json.dumps(out)
        })

    messages += [msg, *tool_results]   # 응답과 결과를 모두 누적
    hop += 1

실무 팁(한 줄씩)

  • while 은 맞는데, 가드레일 이 핵심: MAX_HOPS, per-call timeout/재시도 , 사용자 승인 , 로깅.

  • tool_call_id 매칭 누락/messages 누적 누락이 가장 흔한 버그.

  • 한 턴에 여러 tool_use는 병렬 처리하고, 실패한 것만 선택 재시도.

  • 스트리밍이면 중간에 tool_use 이벤트가 나올 수 있으니 이벤트 파서 로 처리.

결론: 네, “while문으로 도는 반응형 툴-호출 루프”라고 이해하시면 정확합니다.

python에서 AsyncOpenAI 로 client.chat.completions.create 를 통해서 통신을 이미 하고 있거든? 이걸 기반으로 파이썬 코드 한 번 최대한 꼼꼼하게 만들어보자.

좋습니다. 이미 AsyncOpenAI 의 client.chat.completions.create(...)로 모델을 호출하고 계시니, 그 위에 MCP 툴 호출 루프(while loop)를 얹은 프로덕션용 브리지 샘플을 드릴게요.
아래 코드는 다음을 모두 포함합니다.

  • MCP 서버 연결(stdio/HTTP 중 택1)

  • tools/list → OpenAI 함수 포맷으로 변환

  • 모델 응답의 tool_calls 병렬 실행(asyncio.gather)

  • tool_call_id 1:1 매칭으로 tool 메시지 재주입

  • 최대 홉 수, per-call 타임아웃, 재시도, 예외 처리

의존성(예시):
pip install openai anyio + (MCP Python SDK: 보유하신 SDK 이름/버전에 맞춰 mcp 또는 modelcontextprotocol 계열 패키지를 설치하세요)

파이썬: AsyncOpenAI × MCP 브리지 (완성본)

"""
mcp_bridge.py

- AsyncOpenAI Chat Completions 기반의 MCP 브리지
- while 루프: 모델 -> MCP tools/call -> 결과 재주입 -> 종료 조건까지 반복
"""

from __future__ import annotations

import asyncio
import json
import os
import time
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple

from openai import AsyncOpenAI

# === MCP SDK 가져오기 ===
# 사용 중인 MCP Python SDK에 맞게 import 경로를 조정하세요.
# 아래는 가장 흔한 네이밍을 가정한 예시입니다.
try:
    # 예: 공식 MCP Python SDK
    from mcp.client.session import ClientSession
    from mcp.client.stdio import StdioTransport
    # (HTTP 클라이언트가 SDK에 있다면 함께 import)
    try:
        from mcp.client.http import HttpClientTransport  # 있을 때만 사용
    except Exception:
        HttpClientTransport = None  # type: ignore
except Exception as e:
    raise RuntimeError(
        "MCP Python SDK import 실패. 설치 및 경로를 확인하세요.\n"
        "예: pip install modelcontextprotocol  또는  pip install mcp"
    ) from e

# ------------------------------------------------------------
# 데이터 모델
# ------------------------------------------------------------
@dataclass
class McpTool:
    name: str
    description: Optional[str]
    input_schema: Dict[str, Any]  # JSON Schema

# ------------------------------------------------------------
# MCP 클라이언트 래퍼
# ------------------------------------------------------------
class SimpleMcpClient:
    """
    MCP 클라이언트를 간단 래핑. SDK의 실제 인터페이스에 맞게 조정하세요.
    """

    def __init__(
        self,
        stdio_cmd: Optional[List[str]] = None,
        stdio_env: Optional[Dict[str, str]] = None,
        http_url: Optional[str] = None,
        http_headers: Optional[Dict[str, str]] = None,
    ) -> None:
        if stdio_cmd:
            self.transport = StdioTransport(command=stdio_cmd, env=stdio_env or {})
        elif http_url and HttpClientTransport:
            self.transport = HttpClientTransport(url=http_url, headers=http_headers or {})
        else:
            raise ValueError("stdio_cmd 또는 http_url 중 하나는 필요합니다.")
        self.session: Optional[ClientSession] = None

    async def __aenter__(self) -> "SimpleMcpClient":
        self.session = ClientSession(self.transport)
        await self.session.__aenter__()  # 세션/initialize 핸드셰이크
        return self

    async def __aexit__(self, exc_type, exc, tb) -> None:
        if self.session:
            await self.session.__aexit__(exc_type, exc, tb)

    # ---- 기능: tools/list ----
    async def list_tools(self) -> List[McpTool]:
        assert self.session is not None, "MCP 세션이 열려있지 않습니다."
        resp = await self.session.list_tools()
        tools: List[McpTool] = []
        for t in resp.tools:
            tools.append(
                McpTool(
                    name=t.name,
                    description=getattr(t, "description", None),
                    input_schema=getattr(t, "inputSchema", {}) or {},
                )
            )
        return tools

    # ---- 기능: tools/call ----
    async def call_tool(self, name: str, args: Dict[str, Any]) -> Dict[str, Any]:
        """
        MCP tools/call 실행 후 결과를 통일된 dict로 반환합니다.
        SDK마다 resp.content 형태가 다를 수 있으니 여기서 정규화하세요.
        """
        assert self.session is not None, "MCP 세션이 열려있지 않습니다."
        resp = await self.session.call_tool(name=name, arguments=args)

        # content 정규화 예시:
        # resp.content가 [{"type":"text","text":"..."}, {"type":"json","json":{...}}] 형태라고 가정
        out: Dict[str, Any] = {"text": None, "json": None, "raw": None}
        try:
            if hasattr(resp, "content") and isinstance(resp.content, list):
                texts = []
                json_obj = None
                for c in resp.content:
                    if getattr(c, "type", None) == "text":
                        texts.append(getattr(c, "text", ""))
                    elif getattr(c, "type", None) == "json":
                        json_obj = getattr(c, "json", None)
                out["text"] = "\n".join([t for t in texts if t])
                out["json"] = json_obj
                out["raw"] = [c.__dict__ for c in resp.content]  # 디버깅용
            else:
                # 혹시 다른 형태면 통으로 감싸서 반환
                out["raw"] = getattr(resp, "__dict__", resp)
        except Exception:
            out["raw"] = getattr(resp, "__dict__", resp)

        return out

# ------------------------------------------------------------
# 유틸: MCP 툴을 OpenAI tools 포맷으로 변환
# ------------------------------------------------------------
def mcp_tools_to_openai_tools(mcp_tools: List[McpTool]) -> List[Dict[str, Any]]:
    """
    OpenAI Chat Completions `tools` 파라미터 규격:
    [
      {
        "type": "function",
        "function": {
          "name": str,
          "description": str,
          "parameters": {JSON Schema}
        }
      }
    ]
    """
    out: List[Dict[str, Any]] = []
    for t in mcp_tools:
        name = t.name[:64]  # OpenAI 제약: 최대 64자 권장
        params = t.input_schema or {"type": "object", "properties": {}}
        if "type" not in params:
            # 일부 서버가 "object" 생략 시 대비
            params = {"type": "object", **params}
        out.append(
            {
                "type": "function",
                "function": {
                    "name": name,
                    "description": t.description or "",
                    "parameters": params,
                },
            }
        )
    return out

# ------------------------------------------------------------
# 브리지: 모델 <-> MCP 호출 루프
# ------------------------------------------------------------
class McpBridge:
    def __init__(
        self,
        oai: AsyncOpenAI,
        model: str,
        mcp: SimpleMcpClient,
        *,
        max_hops: int = 8,
        per_tool_timeout_s: float = 30.0,
        per_tool_retries: int = 1,
        parallel: bool = True,
        system_prompt: Optional[str] = None,
    ) -> None:
        self.oai = oai
        self.model = model
        self.mcp = mcp
        self.max_hops = max_hops
        self.per_tool_timeout_s = per_tool_timeout_s
        self.per_tool_retries = per_tool_retries
        self.parallel = parallel
        self.system_prompt = system_prompt or (
            "You are a careful assistant. "
            "Call tools only when needed, keep arguments minimal and valid JSON."
        )
        self._cached_openai_tools: Optional[List[Dict[str, Any]]] = None

    async def _ensure_tools(self) -> List[Dict[str, Any]]:
        if self._cached_openai_tools is None:
            mcp_tools = await self.mcp.list_tools()
            self._cached_openai_tools = mcp_tools_to_openai_tools(mcp_tools)
        return self._cached_openai_tools

    async def _call_one_tool_with_retry(
        self, call: Dict[str, Any]
    ) -> Tuple[str, Dict[str, Any]]:
        """
        단일 tool_call 실행 + 재시도/타임아웃.
        return: (tool_call_id, tool_result_message)
        """
        tool_call_id = call["id"]
        func = call["function"]
        name = func["name"]
        args_str = func.get("arguments") or "{}"

        # arguments가 JSON 파싱 불가일 때 관대한 파서
        try:
            args = json.loads(args_str) if isinstance(args_str, str) else args_str
        except Exception:
            # 모델이 문자열을 그대로 넘길 수 있으니 감싼다
            args = {"_": args_str}

        last_err: Optional[Exception] = None
        for attempt in range(self.per_tool_retries + 1):
            try:
                # MCP tools/call에 타임아웃 적용
                out = await asyncio.wait_for(
                    self.mcp.call_tool(name, args),
                    timeout=self.per_tool_timeout_s,
                )
                # OpenAI chat.completions에 넣을 tool 메시지 규격
                tool_msg = {
                    "role": "tool",
                    "tool_call_id": tool_call_id,
                    # 모델이 소비하기 쉽게 문자열 + json 모두 전달
                    "content": json.dumps(out, ensure_ascii=False),
                }
                return tool_call_id, tool_msg
            except Exception as e:
                last_err = e
                if attempt < self.per_tool_retries:
                    await asyncio.sleep(min(2 ** attempt, 2.0))  # 간단 백오프
                    continue
                # 재시도 실패 시 에러 내용을 tool 결과로 전달(모델이 핸들링하게)
                err_payload = {
                    "error": str(e),
                    "type": e.__class__.__name__,
                }
                tool_msg = {
                    "role": "tool",
                    "tool_call_id": tool_call_id,
                    "content": json.dumps(err_payload, ensure_ascii=False),
                }
                return tool_call_id, tool_msg
        # 이론상 도달 불가
        raise last_err or RuntimeError("unknown tool call failure")

    async def run(self, user_input: str) -> str:
        """
        while 루프를 돌며 MCP 툴 호출을 처리하고 최종 답변 텍스트를 반환합니다.
        """
        tools = await self._ensure_tools()

        messages: List[Dict[str, Any]] = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user_input},
        ]

        final_text: str = ""
        for hop in range(self.max_hops):
            # 1) 모델 호출
            res = await self.oai.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=tools,
                tool_choice="auto",
                temperature=0.2,
            )

            msg = res.choices[0].message
            tool_calls = msg.tool_calls or []

            # 2) 툴 호출 없으면 종료
            if not tool_calls:
                final_text = msg.content or ""
                break

            # 3) 툴 호출 실행(동턴 병렬 권장)
            if self.parallel and len(tool_calls) > 1:
                results = await asyncio.gather(
                    *[self._call_one_tool_with_retry(c) for c in tool_calls]
                )
                tool_msgs = [m for _, m in results]
            else:
                tool_msgs = []
                for c in tool_calls:
                    _, m = await self._call_one_tool_with_retry(c)
                    tool_msgs.append(m)

            # 4) 모델에 결과 재주입
            messages.append({"role": msg.role, "content": msg.content, "tool_calls": [
                {
                    "id": c.id,
                    "type": c.type,
                    "function": {
                        "name": c.function.name,
                        "arguments": c.function.arguments,
                    },
                } for c in tool_calls
            ]})
            messages.extend(tool_msgs)

        return final_text

# ------------------------------------------------------------
# 실행 예시
# ------------------------------------------------------------
async def main():
    """
    사용법:
      export OPENAI_API_KEY=dummy
      # GPT-OSS 엔드포인트를 OpenAI 호환으로 띄웠다면:
      export OPENAI_BASE_URL=http://localhost:8000/v1
      python -m mcp_bridge "레포 이슈 요약해줘"
    """
    user_query = os.environ.get("USER_QUERY") or "간단한 테스트 쿼리"
    model = os.environ.get("OSS_MODEL", "gpt-oss-20b")

    # AsyncOpenAI: GPT-OSS의 OpenAI 호환 엔드포인트로 base_url 지정
    client = AsyncOpenAI(
        base_url=os.environ.get("OPENAI_BASE_URL", "http://localhost:8000/v1"),
        api_key=os.environ.get("OPENAI_API_KEY", "dummy"),
        timeout=60.0,
    )

    # MCP 서버 연결: stdio 또는 HTTP 중 하나 택1
    # (1) stdio 예시: 로컬 MCP 서버 바이너리/스크립트를 서브프로세스로 실행
    stdio_cmd = os.environ.get("MCP_STDIO_CMD")
    stdio_args = os.environ.get("MCP_STDIO_ARGS", "")
    stdio = [stdio_cmd, *([a for a in stdio_args.split(" ") if a])] if stdio_cmd else None

    # (2) HTTP 예시: 원격 MCP 서버
    http_url = os.environ.get("MCP_HTTP_URL")  # 예: http://127.0.0.1:3333/mcp
    http_headers = {"Mcp-Session-Id": os.environ.get("MCP_SESSION_ID", "demo-session")}

    # 우선순위: stdio > http
    async with SimpleMcpClient(stdio_cmd=stdio, http_url=http_url, http_headers=http_headers) as mcp:
        bridge = McpBridge(
            oai=client,
            model=model,
            mcp=mcp,
            max_hops=int(os.environ.get("MAX_HOPS", "8")),
            per_tool_timeout_s=float(os.environ.get("TOOL_TIMEOUT_S", "30")),
            per_tool_retries=int(os.environ.get("TOOL_RETRIES", "1")),
            parallel=True,
            system_prompt=(
                "You are a tool-using assistant. "
                "Only call tools when necessary. Keep outputs concise."
            ),
        )
        start = time.time()
        result = await bridge.run(user_query)
        dur = time.time() - start
        print("\n=== FINAL ANSWER ===\n")
        print(result)
        print(f"\n[done in {dur:.2f}s]")

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

어떻게 쓰면 되나? (step-by-step)

  1. GPT-OSS 엔드포인트(OpenAI 호환) 준비

    • OPENAI_BASE_URL를 여러분 엔드포인트로 지정

    • OPENAI_API_KEY는 서버가 요구하면 넣고, 아니면 아무 문자열도 OK

  2. MCP 서버 하나 골라 연결

    • stdio : 로컬 서버를 서브프로세스로 띄우려면 
      arduino
       
export MCP_STDIO_CMD="npx"
export MCP_STDIO_ARGS="@modelcontextprotocol/server-github --stdio"

처럼 커맨드를 환경변수로 넘기세요.

 * **HTTP** : 원격 MCP 서버가 있으면
   ```
   arduino
   
   export MCP_HTTP_URL="http://127.0.0.1:3333/mcp"
   export MCP_SESSION_ID="my-session-1"
   ```
  1. 실행 
    export OPENAI_BASE_URL=http://localhost:8000/v1
export OPENAI_API_KEY=dummy
export OSS_MODEL=gpt-oss-20b
python mcp_bridge.py

또는 ini USER_QUERY="우리 이슈 목록 중 버그만 요약해줘" python mcp_bridge.py

  1. 확장 포인트(현업 팁)

    • McpBridge.run() 안의 temperature, tool_choice, system_prompt를 상황에 맞게 조정

    • per_tool_timeout_s, per_tool_retries로 MCP 안정화

    • 메시지 로깅/감사의 경우, messages 스냅샷과 MCP 호출 로그를 파일/DB에 남기세요

    • 한 턴 내 다건 tool_calls는 이미 병렬 처리. 실패한 것만 부분 재시도하도록 커스터마이즈 가능

    • 필요하면 스트리밍(stream=True)으로 변경하고, 토큰 중간 tool_calls 이벤트를 파싱하는 코루틴을 추가

필요하시면 위 코드를 여러 MCP 서버(예: GitHub/FS/DB)와 붙는 실전 예로 커스터마이즈해서 드릴게요. 현재 쓰시는 MCP 서버 명령/URL 을 알려주시면, stdio_cmd/http_url에 맞춰 바로 작동하는 버전으로 손봐드리겠습니다.

서버 없이 그냥 간단한 파일 조회 같은건 그냥 파이썬 코드로도 할 수 있잖아. 그런 방법은 어때?

맞아요—그냥 로컬 파이썬 함수들을 “툴”로 노출 해서 모델이 부르면 실행해 주면 됩니다. MCP 서버는 건너뛰고, 호스트(브리지) 레벨에서 파일 시스템 전용 툴 을 직접 구현하는 거죠. 간단‧빠름‧의존성 적고, 사내용 파일 탐색/요약엔 충분히 좋습니다.

아래는 여러분이 이미 쓰고 계신 AsyncOpenAI +client.chat.completions.create 위에 얹는, 로컬 파일 툴 브리지(while 루프) 완성 예시예요.

어떻게 동작해?

  1. 툴 정의(JSON Schema) : list_dir, read_text, find_in_files

  2. 모델 호출 → tool_calls 있나 확인

  3. 있으면 해당 파이썬 함수 실행(루트 디렉토리 안에서만, 크기/경로 가드)

  4. tool 결과 메시지 를 다시 모델에 주입

  5. tool_calls가 더 이상 안 나올 때까지 while 반복

보안 가드: 루트 고정(SAFE_ROOT), 경로 정규화/부모 탈출 방지, 파일 크기 제한, 바이너리 차단, 인코딩 폴백(utf-8, utf-8-sig, cp949), 타임아웃, 재시도.

코드 (복사-붙여넣기)

# local_fs_tools_bridge.py
# AsyncOpenAI × 로컬 파일 툴(서버 없음) × while 루프
# pip install openai

from __future__ import annotations
import asyncio, json, os, re
from pathlib import Path
from typing import Any, Dict, List, Optional
from openai import AsyncOpenAI

# === 설정 ===
SAFE_ROOT = Path(os.environ.get("SAFE_ROOT", ".")).resolve()  # 이 디렉토리 안에서만 작동
MAX_BYTES_READ = int(os.environ.get("MAX_BYTES_READ", "65536"))  # 64KB
MAX_LIST_ENTRIES = int(os.environ.get("MAX_LIST_ENTRIES", "200"))
MAX_FILES_SCAN = int(os.environ.get("MAX_FILES_SCAN", "400"))
MODEL = os.environ.get("OSS_MODEL", "gpt-oss-20b")

# === 유틸: 경로 안전성 보장 ===
def _safe_join(rel: str) -> Path:
    """SAFE_ROOT 기준 상대경로를 절대경로로 변환하고, 루트 밖 접근을 차단."""
    p = (SAFE_ROOT / rel).resolve()
    try:
        p.relative_to(SAFE_ROOT)
    except ValueError:
        raise PermissionError(f"Path escapes SAFE_ROOT: {p}")
    return p

def _is_text(path: Path) -> bool:
    """간단한 바이너리 판단: 널 바이트 유무 + 가벼운 디코딩 시도."""
    try:
        with path.open("rb") as f:
            head = f.read(2048)
        if b"\x00" in head:
            return False
        for enc in ("utf-8", "utf-8-sig", "cp949"):
            try:
                head.decode(enc)
                return True
            except Exception:
                continue
        return False
    except Exception:
        return False

# === 로컬 툴 구현 ===
async def tool_list_dir(path: str = ".", glob: str = "*", include_hidden: bool = False) -> Dict[str, Any]:
    base = _safe_join(path)
    if not base.exists():
        return {"error": f"not found: {path}"}
    if base.is_file():
        base = base.parent
    entries = []
    count = 0
    for p in base.glob(glob):
        # 숨김 파일 제외 옵션
        name = p.name
        if not include_hidden and name.startswith("."):
            continue
        # SAFE_ROOT 밖 symlink는 차단
        try:
            resolved = p.resolve()
            resolved.relative_to(SAFE_ROOT)
        except Exception:
            continue
        info = {
            "name": name,
            "path": str(p.relative_to(SAFE_ROOT)),
            "type": "dir" if p.is_dir() else "file",
            "size": (p.stat().st_size if p.is_file() else None),
        }
        entries.append(info)
        count += 1
        if count >= MAX_LIST_ENTRIES:
            break
    return {"root": str(SAFE_ROOT), "dir": str(base.relative_to(SAFE_ROOT)), "count": len(entries), "entries": entries}

async def tool_read_text(path: str, max_bytes: Optional[int] = None) -> Dict[str, Any]:
    maxb = max_bytes or MAX_BYTES_READ
    p = _safe_join(path)
    if not p.exists() or not p.is_file():
        return {"error": f"file not found: {path}"}
    if not _is_text(p):
        return {"error": "binary or non-text file (blocked)"}
    size = p.stat().st_size
    content = ""
    enc_used = None
    # 인코딩 폴백
    for enc in ("utf-8", "utf-8-sig", "cp949"):
        try:
            with p.open("r", encoding=enc, errors="replace") as f:
                content = f.read(maxb + 1)
                enc_used = enc
            break
        except Exception:
            continue
    truncated = len(content) > maxb
    if truncated:
        content = content[:maxb]
    return {
        "path": str(p.relative_to(SAFE_ROOT)),
        "encoding": enc_used,
        "size": size,
        "truncated": truncated,
        "text": content,
    }

async def tool_find_in_files(root: str = ".", pattern: str = "", regex: bool = False,
                             glob: str = "**/*", case_sensitive: bool = False,
                             max_files: Optional[int] = None, max_hits_per_file: int = 5) -> Dict[str, Any]:
    base = _safe_join(root)
    if not base.exists() or not base.is_dir():
        return {"error": f"dir not found: {root}"}
    flags = 0 if case_sensitive else re.IGNORECASE
    rx = re.compile(pattern if regex else re.escape(pattern), flags)
    results = []
    files_scanned = 0
    hit_files = 0
    limit = max_files or MAX_FILES_SCAN

    for p in base.glob(glob):
        if files_scanned >= limit:
            break
        try:
            resolved = p.resolve()
            resolved.relative_to(SAFE_ROOT)
        except Exception:
            continue
        if not p.is_file():
            continue
        if not _is_text(p):
            continue
        files_scanned += 1

        # 크기 제한 내에서만 스캔
        try:
            with p.open("r", encoding="utf-8", errors="replace") as f:
                text = f.read(MAX_BYTES_READ)
        except Exception:
            continue

        hits = []
        for i, line in enumerate(text.splitlines(), start=1):
            if rx.search(line):
                hits.append({"line": i, "preview": line[:300]})
                if len(hits) >= max_hits_per_file:
                    break
        if hits:
            results.append({"path": str(p.relative_to(SAFE_ROOT)), "hits": hits})
            hit_files += 1

    return {
        "root": str(base.relative_to(SAFE_ROOT)),
        "files_scanned": files_scanned,
        "hit_files": hit_files,
        "pattern": pattern,
        "results": results,
        "truncated_note": f"Scanned up to {MAX_BYTES_READ} bytes per file, {limit} files max.",
    }

# === OpenAI tools 스키마 ===
OPENAI_TOOLS: List[Dict[str, Any]] = [
    {
        "type": "function",
        "function": {
            "name": "list_dir",
            "description": "List directory entries under SAFE_ROOT.",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Relative dir path from SAFE_ROOT", "default": "."},
                    "glob": {"type": "string", "description": "glob pattern", "default": "*"},
                    "include_hidden": {"type": "boolean", "default": False},
                },
                "required": []
            }
        },
    },
    {
        "type": "function",
        "function": {
            "name": "read_text",
            "description": "Read a text file with size/encoding safeguards.",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Relative file path from SAFE_ROOT"},
                    "max_bytes": {"type": "integer", "minimum": 1, "maximum": 1048576}
                },
                "required": ["path"]
            }
        },
    },
    {
        "type": "function",
        "function": {
            "name": "find_in_files",
            "description": "Search for a string/regex across text files.",
            "parameters": {
                "type": "object",
                "properties": {
                    "root": {"type": "string", "default": "."},
                    "pattern": {"type": "string", "description": "string or regex"},
                    "regex": {"type": "boolean", "default": False},
                    "glob": {"type": "string", "default": "**/*"},
                    "case_sensitive": {"type": "boolean", "default": False},
                    "max_files": {"type": "integer", "minimum": 1, "maximum": 5000},
                    "max_hits_per_file": {"type": "integer", "minimum": 1, "maximum": 50, "default": 5},
                },
                "required": ["pattern"]
            }
        },
    },
]

# === 라우터 ===
async def call_local_tool(name: str, args: Dict[str, Any]) -> Dict[str, Any]:
    if name == "list_dir":
        return await tool_list_dir(**args)
    if name == "read_text":
        return await tool_read_text(**args)
    if name == "find_in_files":
        return await tool_find_in_files(**args)
    return {"error": f"unknown tool: {name}"}

def _assistant_msg_dict(msg) -> Dict[str, Any]:
    """OpenAI SDK 객체를 Chat API 재호출용 dict로 변환."""
    tool_calls = []
    if msg.tool_calls:
        for c in msg.tool_calls:
            tool_calls.append({
                "id": c.id,
                "type": c.type,
                "function": {
                    "name": c.function.name,
                    "arguments": c.function.arguments,
                },
            })
    return {"role": "assistant", "content": msg.content, "tool_calls": tool_calls or None}

async def run_query(user_input: str) -> str:
    client = AsyncOpenAI(
        base_url=os.environ.get("OPENAI_BASE_URL", "http://localhost:8000/v1"),
        api_key=os.environ.get("OPENAI_API_KEY", "dummy"),
        timeout=60.0,
    )

    system_prompt = (
        "You can inspect files inside SAFE_ROOT only. "
        f"SAFE_ROOT is '{SAFE_ROOT}'. "
        "Call tools only when needed. Prefer concise outputs."
    )
    messages: List[Dict[str, Any]] = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_input},
    ]

    MAX_HOPS = int(os.environ.get("MAX_HOPS", "8"))
    final_text = ""

    for _ in range(MAX_HOPS):
        resp = await client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=OPENAI_TOOLS,
            tool_choice="auto",
            temperature=0.2,
        )
        msg = resp.choices[0].message
        calls = msg.tool_calls or []

        if not calls:  # 종료
            final_text = msg.content or ""
            break

        # 동턴 병렬 실행
        async def _run_one(c):
            name = c.function.name
            try:
                args = json.loads(c.function.arguments or "{}")
            except Exception:
                args = {"_": c.function.arguments}
            out = await asyncio.wait_for(call_local_tool(name, args), timeout=20.0)
            return {
                "role": "tool",
                "tool_call_id": c.id,
                "content": json.dumps(out, ensure_ascii=False)
            }

        tool_msgs = await asyncio.gather(*[_run_one(c) for c in calls])

        # 응답(assistant) + 각 tool 결과를 누적
        messages.append(_assistant_msg_dict(msg))
        messages.extend(tool_msgs)

    return final_text

# === CLI 실행 ===
if __name__ == "__main__":
    import sys
    q = " ".join(sys.argv[1:]) or "data 폴더에서 TODO가 들어간 파일들 찾아줘. 그리고 상위 3개만 요약해."
    print(f"[SAFE_ROOT] {SAFE_ROOT}")
    print(f"[QUERY] {q}")
    print(asyncio.run(run_query(q)))

실행 방법

# 1) 환경변수: 여러분의 OpenAI-호환 GPT-OSS 엔드포인트
export OPENAI_BASE_URL=http://localhost:8000/v1
export OPENAI_API_KEY=dummy
export OSS_MODEL=gpt-oss-20b

# 2) 파일 시스템 루트(이 경로 밖은 접근 금지)
export SAFE_ROOT=/path/to/workspace

# 3) 실행
python local_fs_tools_bridge.py "data 폴더에서 TODO 포함 파일 찾아 요약"

언제 이 방식이 좋은가 / 아닌가

  • 👍 좋음 : 내부 스크립트·리포 정리, 로그 조사, 빌드 산출물 검토 등 “한 머신 안” 단순 오토메이션

  • 🤔 아쉬움 : 여러 앱/팀에서 재사용 하거나, 네트워크 자원/DB/API 같은 원격 리소스 까지 표준화하고 싶다면 MCP 서버화 가 장점(발견성·권한·감사·재사용)

원하시면 위 툴셋에 write_text, tail, zip_list, json_query(jq) 같은 툴을 추가해 드릴게요. 현재 쓰는 폴더 구조나 용례를 알려주시면, 스키마/가드를 맞춰 커스터마이즈해 드립니다.

이걸 표준화된 형태로 할 수는 없을까? harmony도 다 호환되는거야?

맞아요—표준화 하려면 딱 두 축만 잡으면 됩니다:
(1) 툴/데이터 인터페이스 표준 = MCP(서버·클라이언트 규약)
(2) 대화/툴호출 포맷 표준 = Harmony(모델 프롬프트/응답 포맷)
둘은 서로 독립 이지만 잘 맞물립니다. MCP는 “도구를 어떻게 노출·호출할까”를 표준화하고, Harmony는 “모델과의 메시지 포맷”을 표준화해요. modelcontextprotocol.io+1

무엇이 “표준”인가 (한눈에)

  • MCP : JSON-RPC 기반 프로토콜. 표준 전송은 stdio 와 Streamable HTTP 두 가지. 공식 Python/TS SDK가 있고, 여기에 맞춰 툴/리소스를 내보내면 어떤 호스트/에이전트에서도 재사용됩니다. modelcontextprotocol.io+2

  • Harmony : GPT-OSS 계열이 학습한 대화·툴호출 포맷. Responses API 스타일을 모사하며, 공식 렌더러(openai-harmony)가 있어 Python에서도 그대로 씁니다. (여러 배포자/서빙 솔루션은 내부에서 자동 처리하기도 함) GitHub+1

요약: MCP로 도구 측을 표준화 하고, Harmony로 모델 메시지 포맷을 표준화 하면 끝.

내가 가진 “로컬 파이썬 함수”를 표준화하는 두 길

A) 최소 비용으로 MCP 서버 로 감싸기 (권장)

로컬 함수들을 MCP 툴 로 내보내면, Claude/ChatGPT/Gemini/에이전트 SDK에서 그대로 붙일 수 있어요. 파이썬에선 FastMCP 가 가장 손쉬움(데코레이터만 붙이면 schema 생성 포함). GitHub+1

예시(30줄대): 로컬 파일 툴 → MCP 서버

# mcp_files.py
from fastmcp import FastMCP
from pathlib import Path

mcp = FastMCP("LocalFS")
SAFE_ROOT = Path("/path/to/root").resolve()

def safe(rel: str) -> Path:
    p = (SAFE_ROOT / rel).resolve()
    p.relative_to(SAFE_ROOT)  # 밖 탈출 시 예외
    return p

@mcp.tool
def list_dir(path: str = ".") -> list[dict]:
    base = safe(path)
    return [{"name": p.name, "type": "dir" if p.is_dir() else "file"} for p in base.iterdir()]

@mcp.tool
def read_text(path: str, max_bytes: int = 65536) -> dict:
    p = safe(path)
    text = p.read_text(encoding="utf-8", errors="replace")[:max_bytes]
    return {"path": str(p.relative_to(SAFE_ROOT)), "text": text}

if __name__ == "__main__":
    mcp.run()  # stdio (CLI: `fastmcp run mcp_files.py`)

  • 실행: uv pip install fastmcp → fastmcp run mcp_files.py

  • 필요 시 HTTP(스트리머블) 전송으로 바꿔 다중 클라이언트도 처리 가능. GitHub+1

B) 서버 없이, 호스트 안에서 Harmony만 채택

지금처럼 AsyncOpenAI로 while 루프 를 돌리되, 메시지를 Harmony 렌더러 로 구성하면 GPT-OSS 쪽 호환성이 올라갑니다. GPT-OSS는 Harmony 포맷을 전제로 학습되었고, 공식 가이드/렌더러가 있습니다. (단, vLLM/일부 프로바이더는 내부에서 Harmony를 자동 처리하기도 함) GitHub+1

“Harmony도 다 호환되는 거야?”에 대한 정리

  • Harmony ↔ MCP는 층이 다릅니다.
    MCP는 툴·리소스 호출 표준 , Harmony는 대화/툴호출 포맷 표준 이에요. 보통 “호스트”가 MCP에서 얻은 툴 스키마를 모델이 이해하는 포맷(예: Harmony 또는 OpenAI tools 포맷)으로 바꿔 넣습니다. 그래서 동시에 써도 문제 없습니다. openai.github.io

  • GPT-OSS 사용 시 : 직접 서빙하는 엔진이 Harmony를 자동 처리하지 않는다면 , openai-harmony 같은 렌더러 를 쓰세요. 그렇지 않으면 툴 호출/컨텍스트 주입이 불안정할 수 있습니다. OpenAI Cookbook

단계별 권장 루트 (Step-by-Step)

  1. 로컬 함수 → MCP 서버화(FastMCP 추천). 위 예시 같은 파일 툴부터 노출. GitHub

  2. 호스트(브리지)에서 MCP tools/list → 모델용 포맷으로 변환(Harmony 또는 Chat Completions tools).

  3. 툴 호출 루프(while): tool_use 나오면 MCP call_tool 실행 → tool_result 재주입(또는 ChatCompletions의 tool 메시지 규격). ID 1:1 매칭 지키기. modelcontextprotocol.io

  4. 전송·보안 : 로컬은 stdio, 분산은 Streamable HTTP + 세션/인증. modelcontextprotocol.io

필요하시면, 지금 쓰시는 AsyncOpenAI 호출 루프 에 맞춰

  • (A) 위 로컬 파일 툴을 FastMCP 서버 로 감싼 버전,

  • (B) Harmony 렌더러 를 적용한 브리지(while 루프)
    두 가지를 바로 붙여드릴게요.

서버로 감싸면 서버를 또 띄워야하는거아냐?

맞아요, “서버로 감싼다” 해도 꼭 항상 떠 있는 데몬 을 의미하진 않아요.
MCP는 두 전송을 지원하는데:

  • stdio(권장, 로컬) : 호스트가 필요할 때만 서브프로세스 를 띄웁니다. 포트도 없고, 끝나면 같이 내려가요. “또 하나의 인프라”가 아니라 일회성 프로세스 느낌.

  • HTTP(원격/다중 클라) : 진짜 서버를 띄워서 여러 호스트가 붙게 할 때.

즉, 여러분 환경에선 stdio면 충분 해서 “상시 서버 운영” 안 해도 됩니다. while 루프 돌리는 호스트가 알아서 켰다 끄면 끝!

바로 쓰는 패턴 (Step-by-step)

  1. 로컬 툴을 MCP 서버로 래핑 (stdio용, 20~30줄)
# mcp_files.py  — FastMCP 예시 (pip/uv: fastmcp 설치)
from fastmcp import FastMCP
from pathlib import Path

mcp = FastMCP("LocalFS")
SAFE_ROOT = Path(".").resolve()

def safe(rel: str) -> Path:
    p = (SAFE_ROOT / rel).resolve()
    p.relative_to(SAFE_ROOT)  # 밖으로 나가면 예외
    return p

@mcp.tool
def list_dir(path: str = "."):
    base = safe(path)
    return [{"name": p.name, "type": "dir" if p.is_dir() else "file"} for p in base.iterdir()]

@mcp.tool
def read_text(path: str, max_bytes: int = 65536):
    p = safe(path)
    text = p.read_text(encoding="utf-8", errors="replace")[:max_bytes]
    return {"path": str(p.relative_to(SAFE_ROOT)), "text": text}

if __name__ == "__main__":
    mcp.run()  # 기본이 stdio: 호스트가 서브프로세스로 띄움
  1. 호스트에서 “서브프로세스”로 붙이기 (포트 없음)
# host.py — 여러분이 이미 쓰는 AsyncOpenAI while 루프에 이 부분만 추가
import asyncio, json, os
from openai import AsyncOpenAI
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioTransport

async def connect_mcp():
    # 호스트가 필요할 때만 서버 스크립트를 서브프로세스로 실행
    t = StdioTransport(command=["python", "mcp_files.py"])
    sess = ClientSession(t)
    await sess.__aenter__()              # initialize 교섭
    return sess

def mcp_tools_to_openai(tools):
    return [{
        "type":"function",
        "function":{
            "name": t.name,
            "description": getattr(t, "description", "") or "MCP tool",
            "parameters": getattr(t, "inputSchema", {}) or {"type":"object","properties":{}}
        }
    } for t in tools]

async def main():
    client = AsyncOpenAI(
        base_url=os.environ.get("OPENAI_BASE_URL","http://localhost:8000/v1"),
        api_key=os.environ.get("OPENAI_API_KEY","dummy"),
    )
    async with await connect_mcp() as mcp:
        tools = await mcp.list_tools()
        oai_tools = mcp_tools_to_openai(tools)

        messages = [
            {"role":"system","content":"Use tools only when needed."},
            {"role":"user","content":"현재 폴더 목록 보여줘"}
        ]

        # === while 루프 핵심 ===
        for _ in range(6):
            res = await client.chat.completions.create(
                model=os.environ.get("OSS_MODEL","gpt-oss-20b"),
                messages=messages, tools=oai_tools, tool_choice="auto", temperature=0.2
            )
            msg = res.choices[0].message
            if not msg.tool_calls:
                print(msg.content or "")
                break

            # 동턴 tool_calls 실행 후 결과 재주입
            tool_msgs = []
            for c in msg.tool_calls:
                args = json.loads(c.function.arguments or "{}")
                out = await mcp.call_tool(name=c.function.name, arguments=args)
                tool_msgs.append({
                    "role":"tool",
                    "tool_call_id": c.id,
                    "content": json.dumps(
                        {"text": getattr(out, "text", None) if hasattr(out,"text") else None,
                         "raw": getattr(out, "__dict__", str(out))}, ensure_ascii=False)
                })

            # assistant 응답 + tool 결과를 누적
            messages.append({
                "role":"assistant",
                "content": msg.content,
                "tool_calls": [{
                    "id": c.id, "type": c.type,
                    "function": {"name": c.function.name, "arguments": c.function.arguments}
                } for c in msg.tool_calls]
            })
            messages.extend(tool_msgs)

if __name__ == "__main__":
    asyncio.run(main())
  • 실행:
export OPENAI_BASE_URL=http://localhost:8000/v1
export OPENAI_API_KEY=dummy
export OSS_MODEL=gpt-oss-20b
python host.py

여기서 host.py가 mcp_files.py를 그때그때 서브프로세스로 실행 하고, 작업 끝나면 프로세스도 같이 종료됩니다. 네트워크 포트/데몬 X.

언제 진짜 “서버(HTTP)”가 유리한가?

  • 여러 호스트(IDE/에이전트)가 동시에 공유 할 공용 툴일 때

  • 원격/컨테이너/서버리스 등 분산 배치 가 필요할 때

  • 관찰성/모니터링/스케일링 을 표준 인프라로 묶고 싶을 때

그 외엔 stdio 서브프로세스 가 가장 단순하고 안전해요.
정리하면: “서버 또 띄워?” → (대부분) 아니요. 호스트가 잠깐 켰다 끄는 stdio면 충분합니다.

지금 보면 mcp_files.py라는 파일 하나를 mcp로 띄우잖아. 그거 대신에 그냥 함수나 클래스 단위로 올릴수없나?

네, “파일 하나(mcp_files.py)를 띄우는 방식” 말고 함수/클래스 단위로 바로 올리는 표준적인 방법이 있어요. 포인트는 MCP 서버 인스턴스를 만들 때 그 함수·메서드를 등록 해 주는 겁니다. FastMCP 기준으로 두 가지 패턴이 깔끔합니다.

1) 팩토리 함수 엔트리포인트(권장)

별도의 데몬 없이, 호스트가 필요할 때만 stdio 서브프로세스 로 불러 쓰세요. FastMCP는 fastmcp run <모듈:팩토리함수>를 지원합니다. 팩토리 함수가 서버 인스턴스를 만들어 반환 하면 끝! gofastmcp.com

# yourpkg/localfs.py  (여기에 “그냥 쓰던” 함수/클래스가 있다고 가정)
from pathlib import Path

class Files:
    def __init__(self, root: str): self.root = Path(root).resolve()
    def list_dir(self, path: str = ".") -> list[dict]:
        base = (self.root / path).resolve()
        base.relative_to(self.root)  # 루트 밖 접근 차단
        return [{"name": p.name, "type": "dir" if p.is_dir() else "file"} for p in base.iterdir()]

def read_text(root: str, path: str, max_bytes: int = 65536) -> dict:
    p = (Path(root).resolve() / path).resolve()
    p.relative_to(Path(root).resolve())
    text = p.read_text(encoding="utf-8", errors="replace")[:max_bytes]
    return {"path": str(p), "text": text}

# yourpkg/server.py  (“팩토리 함수”로 MCP 서버 구성)
from fastmcp import FastMCP
from .localfs import Files, read_text

SAFE_ROOT = "."

async def create_server() -> FastMCP:
    mcp = FastMCP("LocalFS")
    fs = Files(SAFE_ROOT)

    # 함수/메서드를 “등록”만 하면 MCP 툴이 됩니다
    mcp.tool(fs.list_dir)                  # 인스턴스 메서드(바운드 메서드) 등록
    mcp.tool(lambda path: read_text(SAFE_ROOT, path))  # 루트 고정한 래퍼 함수

    return mcp

실행(항상-켜진 서버 필요 X):

uv pip install fastmcp
fastmcp run yourpkg/server.py:create_server   # STDIO 기본
# 또는 원하시면 HTTP/SSE 전송도 지원
fastmcp run yourpkg/server.py:create_server --transport http --port 3333

  • 이 CLI 엔트리포인트 는 파일 이름이 아니라 함수/변수 레벨로 지목 할 수 있고(예: module:factory_fn), 내부에서 장착한 함수/클래스들이 툴로 노출됩니다. gofastmcp.com

  • stdio 전송은 호스트가 그때그때 서브프로세스 생성/종료 해 쓰는 구조라 “서버 또 띄워야 해?” 부담이 적습니다. 필요하면 Streamable HTTP도 표준이에요. modelcontextprotocol.io+1

2) “클래스 메서드 직접 데코레이트” 대신 “등록 메서드” 패턴

인스턴스/클래스 메서드는 정의 시 데코레이터를 바로 붙이지 말고 , 인스턴스를 만든 뒤 바운드 메서드로 등록 하세요. 그래야 self/cls 파라미터가 스키마에 드러나지 않고 올바르게 노출됩니다. (FastMCP 문서의 권장 패턴) gofastmcp.com

from fastmcp import FastMCP

class Math:
    def add(self, x: int, y: int) -> int: return x + y

mcp = FastMCP("Demo")
math = Math()
mcp.tool(math.add)         # ← 인스턴스 만든 뒤 등록(권장)

# (참고) 정적 메서드는 등록해도 되지만, 인스턴스/클래스 메서드는 위 방식이 안전합니다. :contentReference[oaicite:4]{index=4}

호스트(여러분의 AsyncOpenAI 루프)에서 붙이기

기존 while 루프의 MCP 세션 생성 부분만 stdio 커맨드로 교체 하면 됩니다:

from mcp.client.session import ClientSession
from mcp.client.stdio import StdioTransport

transport = StdioTransport(command=["fastmcp","run","yourpkg/server.py:create_server"])
async with ClientSession(transport) as mcp:
    tools = await mcp.list_tools()
    # ↓ 여기서부터는 이전에 쓰시던 tool loop 동일 (tool_calls -> mcp.call_tool -> 재주입)
  • stdio/HTTP 등 전송은 MCP 표준 이므로, 어떤 호스트에서도 같은 방식으로 연결됩니다. modelcontextprotocol.io+1

정리

  • “파일 하나”가 아니라 함수/클래스 단위로 구성 된 모듈을 팩토리 함수 로 묶어 표준 MCP 서버 를 만들고,

  • stdio 로 필요할 때만 띄워 쓰면 상시 서버 운영 없이 재사용 가능한 표준 툴 체계가 됩니다. gofastmcp.com+1

원하시면 지금 쓰시는 함수/클래스를 기반으로 create_server() 스켈레톤을 바로 만들어 드릴게요—패키지 이름/루트 경로만 알려주세요.

아니 봐봐. 예를 들어 아래와 같이 간단히 운영할 수 없어? python @mcp.tool def list_dir(path: str = "."): # implementation @mcp.tool def read_text(path: str): # implementation async def main(mcp_list): async with AsyncOpenAI(...) as client: client.chat.completions.create("Hello", tools=mcp_list) asyncio.run(main([read_text, list_dir])) 이런 식으로 그냥 바로 넣고 싶어

좋아요—that’s exactly doable.
핵심은 파이썬 함수/메서드 자체를 “도구”로 등록 하고, 그 시그니처로부터 OpenAI tools 스키마 를 만들어 client.chat.completions.create(..., tools=...)에 넘긴 뒤, 응답의 tool_calls를 그 함수에 실행→결과 재주입 하는 작은 러너를 갖추면 됩니다. MCP 서버는 전혀 필요 없어요.

아래 코드는 그대로 복붙해서 돌아가도록 만든 초소형 인라인-툴 브리지 예요.

  • @inline_tool로 함수 등록

  • 타입힌트→JSON Schema 변환(기본 타입들)

  • 동기/비동기 함수 모두 지원

  • while 루프(툴 호출 없어질 때까지) + tool_call_id 매칭

  • 간단한 파일 툴 list_dir, read_text 샘플 포함

# inline_tools_bridge.py
# pip install openai
from __future__ import annotations
import asyncio, inspect, json, os
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, get_origin, get_args

from openai import AsyncOpenAI

# ===============================
# 0) 인라인 툴 등록/스키마 생성기
# ===============================
@dataclass
class _ToolEntry:
    func: Callable
    name: str
    desc: str
    schema: Dict[str, Any]

_REGISTRY: Dict[str, _ToolEntry] = {}

def inline_tool(_fn=None, *, name: Optional[str]=None, description: Optional[str]=None):
    """함수에 달아주면 자동으로 OpenAI tools 스키마가 만들어져 등록됩니다."""
    def decorator(fn: Callable):
        tname = (name or fn.__name__)[:64]
        schema = _build_openai_tool_schema(fn, tname, description or (fn.__doc__ or ""))
        _REGISTRY[tname] = _ToolEntry(fn=fn, name=tname, desc=schema["function"]["description"], schema=schema)
        return fn
    return decorator if _fn is None else decorator(_fn)

def _pytype_to_schema(tp) -> Dict[str, Any]:
    # 아주 가벼운 매핑: 필요 시 확장하세요.
    origin = get_origin(tp)
    args = get_args(tp)
    if tp in (str, None) or origin is None and tp is inspect._empty:
        return {"type": "string"}
    if tp in (int,):
        return {"type": "integer"}
    if tp in (float,):
        return {"type": "number"}
    if tp in (bool,):
        return {"type": "boolean"}
    if origin in (list, List):
        item_schema = _pytype_to_schema(args[0] if args else str)
        return {"type": "array", "items": item_schema}
    if origin in (dict, Dict):
        return {"type": "object"}
    # Optional[T] 처리
    if origin is Optional:
        return _pytype_to_schema(args[0])
    # 그 외엔 문자열로 처리(보수적)
    return {"type": "string"}

def _build_openai_tool_schema(fn: Callable, name: str, desc: str) -> Dict[str, Any]:
    sig = inspect.signature(fn)
    properties: Dict[str, Any] = {}
    required: List[str] = []
    for pname, param in sig.parameters.items():
        ann = param.annotation
        sch = _pytype_to_schema(ann)
        if param.default is not inspect._empty:
            sch["default"] = param.default
        else:
            required.append(pname)
        properties[pname] = sch
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": desc,
            "parameters": {"type": "object", "properties": properties, "required": required},
        },
    }

def compiled_tools() -> List[Dict[str, Any]]:
    return [e.schema for e in _REGISTRY.values()]

# ===================================
# 1) 샘플 인라인 툴(파일 관련, 옵션)
# ===================================
SAFE_ROOT = Path(os.environ.get("SAFE_ROOT", ".")).resolve()

def _safe(rel: str) -> Path:
    p = (SAFE_ROOT / rel).resolve()
    p.relative_to(SAFE_ROOT)  # 밖으로 나가면 예외
    return p

@inline_tool
def list_dir(path: str = ".") -> List[Dict[str, Any]]:
    """List directory entries under SAFE_ROOT."""
    base = _safe(path)
    if not base.exists():
        return []
    out = []
    for ch in base.iterdir():
        out.append({
            "name": ch.name,
            "type": "dir" if ch.is_dir() else "file",
            "size": ch.stat().st_size if ch.is_file() else None
        })
    return out

@inline_tool
def read_text(path: str, max_bytes: int = 65536) -> Dict[str, Any]:
    """Read a text file with size/encoding guards."""
    fp = _safe(path)
    text = fp.read_text(encoding="utf-8", errors="replace")[:max_bytes]
    return {"path": str(fp.relative_to(SAFE_ROOT)), "text": text}

# 예시: 비동기 툴도 OK
@inline_tool
async def slow_count_chars(path: str) -> int:
    """Count characters in a file (simulated slow I/O)."""
    await asyncio.sleep(0.05)
    fp = _safe(path)
    return len(fp.read_text(encoding="utf-8", errors="replace"))

# ===================================
# 2) while 루프 러너 (OpenAI 호환)
# ===================================
def _assistant_msg_dict(msg) -> Dict[str, Any]:
    tc = []
    if msg.tool_calls:
        for c in msg.tool_calls:
            tc.append({
                "id": c.id,
                "type": c.type,
                "function": {"name": c.function.name, "arguments": c.function.arguments},
            })
    return {"role": "assistant", "content": msg.content, "tool_calls": tc or None}

async def _call_inline_tool(tool_name: str, args_json: str, timeout_s: float = 20.0) -> Dict[str, Any]:
    if tool_name not in _REGISTRY:
        return {"error": f"unknown tool: {tool_name}"}
    entry = _REGISTRY[tool_name]
    try:
        args = json.loads(args_json or "{}")
    except Exception:
        args = {"_": args_json}

    async def _runner():
        res = entry.func(**args)
        if inspect.iscoroutine(res):
            res = await res
        return res

    try:
        result = await asyncio.wait_for(_runner(), timeout=timeout_s)
        return {"ok": True, "result": result}
    except Exception as e:
        return {"ok": False, "error": str(e), "type": e.__class__.__name__}

async def chat_with_inline_tools(user_input: str) -> str:
    client = AsyncOpenAI(
        base_url=os.environ.get("OPENAI_BASE_URL", "http://localhost:8000/v1"),
        api_key=os.environ.get("OPENAI_API_KEY", "dummy"),
        timeout=60.0,
    )
    model = os.environ.get("OSS_MODEL", "gpt-oss-20b")
    tools = compiled_tools()

    messages: List[Dict[str, Any]] = [
        {"role": "system", "content": f"You can use tools. SAFE_ROOT is '{SAFE_ROOT}'."},
        {"role": "user", "content": user_input},
    ]
    MAX_HOPS = int(os.environ.get("MAX_HOPS", "8"))
    final = ""

    for _ in range(MAX_HOPS):
        resp = await client.chat.completions.create(
            model=model,
            messages=messages,
            tools=tools,
            tool_choice="auto",
            temperature=0.2,
        )
        msg = resp.choices[0].message
        calls = msg.tool_calls or []
        if not calls:
            final = msg.content or ""
            break

        # 동턴 병렬 실행
        async def _one(c):
            out = await _call_inline_tool(c.function.name, c.function.arguments)
            return {"role": "tool", "tool_call_id": c.id, "content": json.dumps(out, ensure_ascii=False)}

        tool_msgs = await asyncio.gather(*[_one(c) for c in calls])

        messages.append(_assistant_msg_dict(msg))
        messages.extend(tool_msgs)

    return final

# ==========================
# 3) 데모 실행
# ==========================
if __name__ == "__main__":
    import sys
    q = " ".join(sys.argv[1:]) or "현재 디렉토리 목록을 보여주고, README.md 있으면 읽어줘."
    print(f"[SAFE_ROOT] {SAFE_ROOT}")
    print(asyncio.run(chat_with_inline_tools(q)))

이렇게 쓰시면 됩니다

export OPENAI_BASE_URL=http://localhost:8000/v1
export OPENAI_API_KEY=dummy
export OSS_MODEL=gpt-oss-20b
export SAFE_ROOT=/your/workspace
python inline_tools_bridge.py "data 폴더 목록 보여주고, README.md 읽어줘"

한 줄 요약 (컨셉 정리)

  • 지금 보여드린 방식은 MCP 없이 “함수/클래스 → OpenAI tools”로 직접 매핑하는 인라인 툴 입니다.

  • 장점 : 초간단, 서버/프로세스 추가 없음.

  • 주의 : 이것만으로는 “MCP 표준”은 아니므로, 다른 호스트/에이전트와 재사용성·발견성 이 필요해지면 같은 함수들을 FastMCP로 래핑 해 stdio로 띄우면 바로 표준화를 얻을 수 있어요(코드 변경 최소).

FastMCP에는 이런 기능이 없어?

짧게 결론부터요:

  • FastMCP 자체로 “파이썬 함수 리스트를 곧장 Chat Completions의 tools에 넣어주는” 내장 기능은 없습니다. FastMCP는 기본적으로 MCP 서버/클라이언트 를 만들기 위한 프레임워크예요. (@mcp.tool로 함수 등록 → MCP로 노출) GitHub+1

  • 다만 선택지는 3가지입니다. (step-by-step)

  1. OpenAI Responses API로 전환
    OpenAI의 Responses API 는 MCP 서버를 “원격 툴 소스”로 바로 붙일 수 있게 표준 지원합니다(※ Chat Completions는 미지원). FastMCP로 HTTP/SSE 서버를 띄우고 Responses에서 type: "mcp"로 연결하면 끝. gofastmcp.com

  2. 서버 따로 띄우기 싫다면: FastMCP “인메모리”로 내장 + 얇은 어댑터
    FastMCP에는 클라이언트가 FastMCP 서버 인스턴스에 직접 붙는 “인메모리 트랜스포트”가 있어, 별도 프로세스 없이 같은 파이썬 프로세스 안에서 MCP 호출을 수행할 수 있어요. 이때 client.list_tools()로 툴 스키마를 읽어 OpenAI 함수-툴 포맷으로 변환 해서 chat.completions.create(..., tools=...)에 넘기고, tool_calls가 오면 client.call_tool(...)로 실행→결과를 재주입하면 됩니다. (아래 소스 15~20줄 정도면 충분) gofastmcp.com

  3. 필요하면 stdio/HTTP로 “그때그때” 구동
    fastmcp run server.py[:객체/팩토리]로 STDIO(기본)/HTTP/SSE 중 아무거나 택해 일회성으로 띄우는 패턴도 공식 제공돼요. (항상-켜진 서버 운영 부담 ↓) GitHub+1

최소 어댑터 예시 (인메모리 FastMCP ↔ Chat Completions)

핵심: FastMCP로 함수 등록 → Client(mcp)로 인메모리 접속 → list_tools()를 OpenAI tools 포맷 으로 변환 → tool_calls는 client.call_tool()로 실행. ```python import asyncio, json, os from fastmcp import FastMCP, Client from openai import AsyncOpenAI

mcp = FastMCP("LocalTools")

@mcp.tool def list_dir(path: str = ".") -> list[dict]: """List directory entries.""" import os return [{"name": f, "type": ("dir" if os.path.isdir(os.path.join(path, f)) else "file")} for f in os.listdir(path)]

@mcp.tool def read_text(path: str, max_bytes: int = 65536) -> dict: """Read a UTF-8 text file.""" with open(path, "r", encoding="utf-8", errors="replace") as f: return {"path": path, "text": f.read()[:max_bytes]}

async def main(user_input: str): # 1) 인메모리로 FastMCP 서버에 붙기(별도 프로세스 X) mcp_client = Client(mcp) async with mcp_client: # 2) MCP 툴 → OpenAI 함수-툴 포맷 변환 tools = await mcp_client.list_tools() openai_tools = [{ "type": "function", "function": { "name": t.name, "description": t.description or "", "parameters": t.inputSchema # MCP 표준 스키마 그대로 사용 } } for t in tools]

    # 3) 일반 Chat Completions 루프
    llm = AsyncOpenAI(
        base_url=os.getenv("OPENAI_BASE_URL", "http://localhost:8000/v1"),
        api_key=os.getenv("OPENAI_API_KEY", "dummy"),
    )
    model = os.getenv("OSS_MODEL", "gpt-oss-20b")
    messages = [{"role": "user", "content": user_input}]

    for _ in range(8):
        resp = await llm.chat.completions.create(
            model=model, messages=messages, tools=openai_tools, tool_choice="auto", temperature=0.2
        )
        msg = resp.choices[0].message
        if not msg.tool_calls:
            print(msg.content or "")
            return

        # 4) tool_calls 실행 → 결과 재주입
        tool_msgs = []
        for c in msg.tool_calls:
            args = json.loads(c.function.arguments or "{}")
            result = await mcp_client.call_tool(c.function.name, args)
            # 결과는 문자열로 넣어주면 안전
            payload = getattr(result, "data", None) or getattr(result, "content", None) or result
            tool_msgs.append({"role": "tool", "tool_call_id": c.id,
                              "content": json.dumps(payload, ensure_ascii=False, default=str)})

        messages.append({
            "role": "assistant",
            "content": msg.content,
            "tool_calls": [{"id": c.id, "type": c.type,
                            "function": {"name": c.function.name, "arguments": c.function.arguments}}
                           for c in msg.tool_calls]
        })
        messages.extend(tool_msgs)

if name == "main": asyncio.run(main("현재 폴더 목록을 보여주고 README.md가 있으면 읽어줘"))


  * 위 코드는 **서버 프로세스 없이** FastMCP를 _내장_ 하고, MCP 툴 스키마를 그대로 **OpenAI 함수-툴** 로 넘깁니다. (Responses API로 갈 수 없다면 이 방식이 가장 가벼움)

  * 인스턴스/클래스 메서드를 툴로 내고 싶다면, **인스턴스를 만들고 바운드 메서드를 등록** 하는 패턴이 권장입니다. (`@mcp.tool`를 메서드에 직접 붙이지 말 것) [gofastmcp.com](https://gofastmcp.com/patterns/decorating-methods)

* * *

## 언제 무엇을 쓰면 되나 (요약)

  * **OpenAI Responses API 사용 가능** → _그냥 FastMCP 서버_ 노출하고 `type: "mcp"`로 붙이세요. (정석) [gofastmcp.com](https://gofastmcp.com/integrations/openai)

  * **Chat Completions만 반드시 써야 함** → 위 **인메모리 + 변환 어댑터** 패턴. (프로세스 추가 無) [gofastmcp.com](https://gofastmcp.com/clients/client)

  * **원하면 일회성 서버** → `fastmcp run server.py[:obj]`로 **STDIO/HTTP/SSE** 중 선택해 띄우기. [GitHub+1](https://github.com/jlowin/fastmcp)

필요하시면, 지금 쓰시는 함수들로 **인메모리 FastMCP + 변환 어댑터** 를 바로 맞춰 드릴게요—함수 시그니처만 알려 주세요.

<p class="chat--user" markdown="1">

fastmcp 깃허브 보니 mcp = FastMCP("My MCP") async def main(): async with Client(mcp) as client: ... 이렇게 쓸 수 있는 것 같은데?

</p>

맞아요, 그 패턴 그대로 쓸 수 있어요. **FastMCP는 “인메모리 클라이언트”**를 제공해서, 별도 서버나 서브프로세스 없이 **같은 파이썬 프로세스 안에서** MCP 호출을 돌릴 수 있습니다. 즉:
```python
mcp = FastMCP("My MCP")

async def main():
    async with Client(mcp) as client:
        ...

이 형태가 공식적으로 안내되는 인메모리(프로세스 내) 연결 방식이고, 테스트·개발에 특히 적합해요. gofastmcp.com+2

아래는 당신이 원하신 그대로, 함수들을 바로 등록 하고 AsyncOpenAI의 chat.completions.create(...)에 툴 스키마를 넣어 while 루프를 돌리는 완성 예시 입니다. (서버/포트/데몬 없음)

fastmcp × 인메모리 × Chat Completions 브리지 (복붙용)

# fastmcp_inline_bridge.py
# pip install fastmcp openai

from __future__ import annotations
import asyncio, json, os, inspect
from typing import Any, Dict, List, Optional
from openai import AsyncOpenAI
from fastmcp import FastMCP, Client

# 0) MCP 인스턴스와 "그냥 함수"들을 MCP 툴로 등록
mcp = FastMCP("LocalFS")

@mcp.tool
def list_dir(path: str = ".") -> list[dict]:
    """List directory entries."""
    import os, pathlib
    p = pathlib.Path(path).resolve()
    entries = []
    for name in os.listdir(p):
        full = p / name
        entries.append({
            "name": name,
            "type": "dir" if full.is_dir() else "file",
            "size": full.stat().st_size if full.is_file() else None
        })
    return entries

@mcp.tool
def read_text(path: str, max_bytes: int = 65536) -> dict:
    """Read a UTF-8 text file (size-limited)."""
    from pathlib import Path
    fp = Path(path).resolve()
    text = fp.read_text(encoding="utf-8", errors="replace")[:max_bytes]
    return {"path": str(fp), "text": text}

# 1) MCP -> OpenAI tools 포맷 변환 도우미
def _schema_of(tool_obj) -> Dict[str, Any]:
    # FastMCP가 내주는 속성 이름이 구현별로 다를 수 있어 대비
    return (
        getattr(tool_obj, "input_schema", None) or
        getattr(tool_obj, "inputSchema", None) or
        {"type": "object", "properties": {}}
    )

def _to_openai_tools(mcp_tools) -> List[Dict[str, Any]]:
    out = []
    for t in mcp_tools:
        out.append({
            "type": "function",
            "function": {
                "name": t.name[:64],
                "description": getattr(t, "description", "") or "MCP tool",
                "parameters": _schema_of(t),
            }
        })
    return out

def _assistant_msg_dict(msg) -> Dict[str, Any]:
    tool_calls = []
    if msg.tool_calls:
        for c in msg.tool_calls:
            tool_calls.append({
                "id": c.id,
                "type": c.type,
                "function": {"name": c.function.name, "arguments": c.function.arguments}
            })
    return {"role": "assistant", "content": msg.content, "tool_calls": tool_calls or None}

async def _call_tool(client: Client, tool_name: str, args_json: str, timeout_s: float = 20.0) -> Dict[str, Any]:
    try:
        args = json.loads(args_json or "{}")
    except Exception:
        args = {"_": args_json}

    async def _runner():
        res = await client.call_tool(tool_name, args)
        # 결과 정규화(문자/객체 무엇이든 JSON으로 넘길 수 있게)
        if hasattr(res, "content"):
            payload = res.content
        elif hasattr(res, "data"):
            payload = res.data
        else:
            payload = res
        try:
            json.dumps(payload, default=str)  # 직렬화 가능 확인
            return {"ok": True, "result": payload}
        except Exception:
            return {"ok": True, "result": str(payload)}

    try:
        return await asyncio.wait_for(_runner(), timeout=timeout_s)
    except Exception as e:
        return {"ok": False, "error": str(e), "type": e.__class__.__name__}

# 2) 메인 루프: Chat Completions ↔ FastMCP(인메모리)
async def chat_with_inline_mcp(user_input: str) -> str:
    # OpenAI-호환 엔드포인트(GPT-OSS 등)로 교체
    oai = AsyncOpenAI(
        base_url=os.getenv("OPENAI_BASE_URL", "http://localhost:8000/v1"),
        api_key=os.getenv("OPENAI_API_KEY", "dummy"),
        timeout=60.0,
    )
    model = os.getenv("OSS_MODEL", "gpt-oss-20b")

    async with Client(mcp) as mcp_client:
        tools = await mcp_client.list_tools()
        openai_tools = _to_openai_tools(tools)

        messages: List[Dict[str, Any]] = [
            {"role": "system", "content": "You can use tools when necessary. Keep outputs concise."},
            {"role": "user", "content": user_input},
        ]

        MAX_HOPS = int(os.getenv("MAX_HOPS", "8"))
        final_text = ""

        for _ in range(MAX_HOPS):
            resp = await oai.chat.completions.create(
                model=model,
                messages=messages,
                tools=openai_tools,
                tool_choice="auto",
                temperature=0.2,
            )
            msg = resp.choices[0].message
            if not msg.tool_calls:           # 종료 조건
                final_text = msg.content or ""
                break

            # 동턴 병렬 실행
            async def _one(c):
                out = await _call_tool(mcp_client, c.function.name, c.function.arguments)
                return {"role": "tool", "tool_call_id": c.id, "content": json.dumps(out, ensure_ascii=False)}

            tool_msgs = await asyncio.gather(*[_one(c) for c in msg.tool_calls])

            # assistant 응답 + tool 결과를 누적
            messages.append(_assistant_msg_dict(msg))
            messages.extend(tool_msgs)

        return final_text

# 3) 실행
if __name__ == "__main__":
    import sys
    q = " ".join(sys.argv[1:]) or "현재 디렉토리 목록 보여주고 README.md 있으면 읽어줘."
    print(asyncio.run(chat_with_inline_mcp(q)))

사용법

pip install fastmcp openai
export OPENAI_BASE_URL=http://localhost:8000/v1
export OPENAI_API_KEY=dummy
export OSS_MODEL=gpt-oss-20b
python fastmcp_inline_bridge.py "data 폴더 목록과 README.md 읽어줘"

정리 (Step-by-step)

  1. @mcp.tool로 그냥 파이썬 함수 를 등록한다.

  2. async with Client(mcp)로 인메모리 연결 을 연다 → list_tools()로 툴 스키마 획득. gofastmcp.com+1

  3. 그 스키마를 OpenAI tools 포맷 으로 변환해 chat.completions.create(..., tools=...)에 전달.

  4. 응답의 tool_calls를 client.call_tool(...)로 실행하고, tool_call_id 매칭하여 tool 메시지 로 재주입 → while 반복.

  5. 별도 서버나 포트는 전혀 필요 없음(인메모리). 필요해지면 언제든 STDIO/HTTP 로 바꿀 수 있음. gofastmcp.com

원하시면 여기에 클래스 메서드 바운드 등록 , 타임아웃/재시도/로깅 까지 커스터마이즈한 실전 버전으로 다듬어 드릴게요.

그런데 예를 들어 이렇게 만들 수 있어? def add_read_text_tool(mcp: FastMCP): # mcp에 read_text를 tool로 등록 후 return mcp my_mcp = FastMCP("My MCP") my_mcp = add_read_text_tool(my_mcp) # my_mcp에는 툴이 등록된 상태

네, 그렇게 만들 수 있어요. 포인트는 함수 자체에는 데코레이터를 붙이지 않고 , 나중에 mcp.tool(fn)으로 등록(attach) 하는 거예요. 그러면 “추가 함수”들을 손쉽게 끼워넣고 조합할 수 있습니다.

아래는 바로 복붙해서 쓰는 유틸 패턴 이에요.

1) 툴 추가 유틸 (함수/클래스 메서드 모두 지원)

# tools_adders.py
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict
from fastmcp import FastMCP

def add_read_text_tool(mcp: FastMCP, *, root: str | Path = ".",
                       name: str = "read_text", default_max_bytes: int = 65536) -> FastMCP:
    """read_text(path, max_bytes) 툴을 mcp에 등록하고 mcp를 반환."""
    ROOT = Path(root).resolve()

    def _safe(rel: str) -> Path:
        p = (ROOT / rel).resolve()
        p.relative_to(ROOT)  # 밖으로 나가면 예외
        return p

    def read_text(path: str, max_bytes: int = default_max_bytes) -> Dict[str, Any]:
        """Read a UTF-8 text file under ROOT with size limit."""
        fp = _safe(path)
        txt = fp.read_text(encoding="utf-8", errors="replace")[:max_bytes]
        return {"path": str(fp.relative_to(ROOT)), "text": txt}

    # 함수 이름을 바꾸고 싶으면 wrapper의 __name__을 바꿔도 되고…
    read_text.__name__ = name
    mcp.tool(read_text)          # ← 여기서 등록!
    return mcp

def add_list_dir_tool(mcp: FastMCP, *, root: str | Path = ".",
                      name: str = "list_dir") -> FastMCP:
    """list_dir(path='.') 툴 등록."""
    ROOT = Path(root).resolve()

    def _safe(rel: str) -> Path:
        p = (ROOT / rel).resolve()
        p.relative_to(ROOT)
        return p

    def list_dir(path: str = ".") -> list[dict]:
        """List directory entries under ROOT."""
        base = _safe(path)
        return [{
            "name": p.name,
            "type": "dir" if p.is_dir() else "file",
            "size": p.stat().st_size if p.is_file() else None
        } for p in base.iterdir()]

    list_dir.__name__ = name
    mcp.tool(list_dir)
    return mcp

2) 사용 예 — 체이닝해서 등록

# main.py
from fastmcp import FastMCP, Client
from tools_adders import add_read_text_tool, add_list_dir_tool

mcp = FastMCP("My MCP")
mcp = add_read_text_tool(mcp, root=".", name="read_text")
mcp = add_list_dir_tool(mcp, root=".", name="list_dir")

# 인메모리로 붙어서 동작 확인
import asyncio, json

async def demo():
    async with Client(mcp) as client:
        tools = await client.list_tools()
        print("== Registered tools ==")
        for t in tools:
            print("-", t.name)

        # 실제 호출
        out = await client.call_tool("list_dir", {"path": "."})
        print("list_dir ->", out)

        out = await client.call_tool("read_text", {"path": "README.md", "max_bytes": 200})
        print("read_text ->", (out["text"][:80] + "...") if "text" in out else out)

asyncio.run(demo())

3) (옵션) 클래스 메서드도 같은 방식으로

from fastmcp import FastMCP

class Files:
    def __init__(self, root: str): self.root = Path(root).resolve()
    def _safe(self, rel: str) -> Path:
        p = (self.root / rel).resolve(); p.relative_to(self.root); return p
    def read_text(self, path: str, max_bytes: int = 65536) -> dict:
        fp = self._safe(path); return {"path": str(fp), "text": fp.read_text()[:max_bytes]}

def add_files_tools(mcp: FastMCP, fs: Files) -> FastMCP:
    mcp.tool(fs.read_text)   # 바운드 메서드 등록 권장(데코레이터를 메서드에 직접 붙이지 말 것)
    return mcp

한 걸음 더 (툴을 LLM에 넘길 때)

위처럼 등록된 툴 스키마는 Client(mcp).list_tools()로 얻어서 OpenAI tools 포맷 으로 변환해 AsyncOpenAI(...).chat.completions.create(..., tools=...)에 넣으면 됩니다(이미 쓰시던 while 루프 그대로). 핵심 단계:

  1. async with Client(mcp) as client:

  2. tools = await client.list_tools()

  3. 각 t에 대해 {"type":"function","function":{"name": t.name, "parameters": t.inputSchema}}로 변환

  4. LLM 호출 → tool_calls 오면 client.call_tool(tname, args) 실행 → 결과를 tool 메시지로 재주입 → 반복

정리

  • 네, 질문하신 “add 함수로 FastMCP 인스턴스에 툴을 주입하고 그대로 반환”_* 패턴이 깔끔하게 동작합니다.

  • 이 방식은 컴포저블(여러 add_* 조합), 테스트 용이 , 프로세스 추가 無(인메모리)라는 장점이 있어 운영에 딱 좋아요.

Categories:

Updated:

You may also enjoy