Novita AI로 첫 MCP 서버 구축하기

Novita AI로 첫 MCP 서버 구축하기

Model Context Protocol(MCP)은 AI 세계에서 가장 주목받는 개념 중 하나로 빠르게 자리 잡고 있습니다. 비교적 새로운 개념임에도 불구하고, 이미 개발자와 주요 기술 기업들의 관심을 사로잡았습니다.

Anthropic이 만든 MCP는 HTTP와 같은 다른 프로토콜과 유사한 개방형 표준 프로토콜입니다. 하지만 HTTP가 서버와 웹 브라우저(또는 HTTP 클라이언트)를 통해 사용자를 리소스에 연결하도록 설계된 반면, MCP는 대규모 언어 모델(LLM)을 외부 도구 및 데이터에 연결하도록 설계되었습니다.

예를 들어, 챗봇이 GitHub에 접근할 수 있기를 원한다고 가정해 보겠습니다. 사용자가 챗봇을 사용하여 이슈를 생성하고, 이슈를 닫고, 댓글을 추가하거나, 기존 풀 리퀘스트를 조회할 수 있도록 하려는 것입니다. 챗봇이 이러한 모든 작업을 수행할 수 있으려면 도구에 대한 접근 권한이 필요합니다.

바로 여기서 MCP 서버가 등장합니다.

MCP 서버는 챗봇이 내부 LLM을 사용하여 호출할 수 있는 여러 도구를 호스팅하여 이러한 기능에 접근할 수 있게 합니다. GitHub의 경우 공식 GitHub MCP 서버를 사용하여 위에서 언급한 모든 작업을 수행할 수 있습니다.

이 글에서는 Novita API를 사용한 MCP 서버를 구축할 것입니다. 이 MCP 서버를 사용하면 MCP 프로토콜을 지원하는 모든 애플리케이션이 Novita에서 제공하는 모든 LLM에 접근하고, 이미지를 생성하고, 비디오를 만들고, 음성 합성을 수행할 수 있습니다.

MCP 아키텍처 이해하기

MCP 아키텍처

MCP 아키텍처 [출처]

MCP 아키텍처는 일반적인 클라이언트-서버 설정과 유사합니다. 네 가지 중요한 구성 요소가 있습니다:

  • MCP 클라이언트
  • MCP 서버
  • MCP 전송
  • MCP 호스트

MCP 클라이언트

MCP 클라이언트는 AI 애플리케이션과 MCP 서버 간의 통신 게이트웨이 역할을 합니다. 이 통신이 설정되면 애플리케이션은 서버가 보유한 모든 도구와 리소스에 접근할 수 있습니다.

MCP 서버

MCP 서버는 클라이언트가 AI 애플리케이션에 제공할 수 있는 모든 도구와 데이터를 호스팅합니다. 다음을 호스팅합니다:

  • 도구
  • 리소스
  • 프롬프트

도구

도구는 LLM이 일반적으로 가지고 있지 않은 외부 기능(예: 시간 알림, 데이터베이스 업데이트 및 읽기, 날씨 가져오기 등)을 제공합니다. 이러한 도구는 기본적으로 프로그래머가 정의한 호출 가능한 함수입니다.

MCP 서버는 여러 도구를 호스팅할 수 있습니다. 이러한 도구는 LLM이 수행할 수 있는 작업 덕분에 에이전트로서 기능할 수 있게 합니다.

도구는 에이전트가 관리하며 사용 중에 선택적인 인간 감독이 포함될 수 있습니다. HTTP 프로토콜의 POST 요청과 유사하게 부작용을 수행합니다.

리소스

리소스는 AI 애플리케이션에 데이터를 제공하는 읽기 전용 정보입니다. HTTP GET 요청과 유사하며 부작용을 일으키지 않습니다. 리소스는 애플리케이션이 관리하며, 사용자나 에이전트가 리소스와 상호 작용하는 방식을 결정합니다. 리소스의 예로는 파일 내용, API 응답, 데이터베이스 레코드 등이 있습니다.

프롬프트

MCP 서버는 프롬프트 템플릿을 호스팅할 수 있습니다. 이러한 프롬프트를 통해 사용자는 MCP 서버에서 프롬프트를 가져와 AI 애플리케이션에서 사용할 수 있습니다. 프롬프트는 사용자가 관리하며, 어떤 프롬프트를 에이전트에 제공할지 결정합니다.

이 모든 것이 MCP 서버를 구성합니다. 이 글에서는 MCP 서버에서 가장 널리 사용되는 부분인 도구만 구현할 것입니다.

MCP 전송

MCP 전송은 MCP 클라이언트가 MCP 서버와 통신하는 방법을 의미합니다. 이 통신은 클라이언트와 서버가 동일한 머신에서 실행될 때 로컬로 발생하거나, 별도 장치에 있을 때 원격으로 발생할 수 있습니다. MCP는 현재 두 가지 주요 전송 메커니즘을 지원합니다:

  • STDIO: 이 모드에서는 MCP 서버와 클라이언트가 동일한 머신에서 실행되며 표준 입력과 출력을 통해 통신합니다.
  • SSE: 이 모드에서는 MCP 서버가 HTTP를 통해 실행됩니다. HTTP POST는 서버에 메시지를 보내는 데 사용되고, Server-Sent Events는 서버에서 클라이언트로 메시지를 보내는 데 사용됩니다.
  • Streamable HTTP: 이 모드도 HTTP를 사용하지만 HTTP GET 및 POST 요청에 의존합니다. 서버에서 클라이언트로 여러 메시지를 스트리밍해야 할 때만 SSE로 폴백합니다.

표준 입력/출력

STDIO를 사용한 MCP 전송

STDIO를 사용한 MCP 전송 [출처]

캡션: STDIO를 사용한 MCP 전송 [출처]

MCP 클라이언트와 서버는 표준 입력과 출력(STDIO)을 통해 통신할 수 있습니다. STDIO를 사용할 때는 클라이언트와 서버가 동일한 머신에서 실행됩니다. 클라이언트는 모든 요청을 stdin에 쓰고, 서버는 응답을 stdout에 씁니다.

Streamable HTTP

Streamable HTTP를 사용한 MCP 전송

Streamable HTTP를 사용한 MCP 전송 [출처]

Streamable HTTP는 MCP 클라이언트와 서버가 HTTP를 통해 통신할 수 있게 합니다. HTTP POST 메서드를 사용하여 클라이언트에서 요청을 보내고 서버에서 응답을 받습니다. 필요한 경우 선택적으로 Server-Sent Events(SSE)로 전환하여 서버에서 클라이언트로 메시지를 스트리밍할 수 있습니다.

Streamable HTTP는 클라이언트와 서버 간의 원격 통신에 적합하며, 더 이상 사용되지 않는 MCP SSE 전송을 대체합니다.

MCP 호스트

MCP가 일반적인 클라이언트-서버 아키텍처와 다른 점은 호스트의 역할입니다. MCP 사양은 이를 클라이언트-호스트-서버 아키텍처로 정의합니다. 클라이언트와 호스트가 아키텍처의 프런트엔드를 구성하기 때문입니다. MCP 호스트는 두 가지 중요한 구성 요소로 이루어져 있습니다:

  • MCP 클라이언트
  • 대규모 언어 모델

MCP 클라이언트의 역할은 LLM이 서버에서 필요한 도구, 리소스 및 프롬프트를 가져오는 것입니다. 리소스가 수집되면 모델의 컨텍스트 내에 배치됩니다. 단일 MCP 호스트는 여러 클라이언트를 가질 수 있으며, 각각 고유한 MCP 서버에 연결됩니다.

MCP 호스트는 또한 사용자가 작업 중인 다른 기능을 수행할 수 있는 애플리케이션으로 간주될 수 있습니다. 예를 들어, Claude Desktop은 챗봇으로 기능하는 MCP 호스트이고, Cursor는 IDE 역할도 하는 MCP 호스트이며, Claude Code는 AI 코딩 에이전트로 설계된 MCP 호스트입니다.

MCP 서버 작업하기

MCP 서버를 구축하기 전에 기존 MCP 서버로 작업하는 방법을 살펴보겠습니다. Python MCP SDK를 사용하여 novita-mcp-server와 상호 작용할 것입니다.

이 글을 작성하는 시점에서 novita-mcp-server는 다음 도구를 제공합니다:

  • 모든 Novita 클러스터를 나열하는 도구
  • Novita GPU 인스턴스 제품을 나열하는 도구
  • 실행 중인 모든 GPU 인스턴스를 나열하는 도구
  • 새 GPU 인스턴스를 생성하는 도구

STDIO를 통해 MCP 서버에 연결하고 서버에서 사용 가능한 모든 도구를 나열하는 간단한 스크립트를 작성하여 이를 확인해 보겠습니다.

먼저 MCP SDK를 설치합니다: pip install “mcp[cli]”

SDK를 설치한 후 client.py 파일을 만듭니다. 그런 다음 필요한 모든 가져오기를 수행합니다:

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import os

그런 다음 StdioServerParameters를 사용하여 stdio를 통해 MCP 서버를 시작하기 위한 매개변수를 설정합니다. novita-mcp-server는 Node.js로 구현되었으므로 사용하려면 npx 명령으로 실행해야 합니다. 또한 Novita API 키를 환경 변수 NOVITA_API_KEY에 저장해야 합니다.

# Create server parameters for stdio connection
server_params = StdioServerParameters(
   command="npx",
   args=["-y", "@novitalabs/novita-mcp-server"],
   env={"NOVITA_API_KEY":  os.environ["NOVITA_API_KEY"]},
)

다음으로 비동기 함수를 만듭니다. 함수 내에서 서버 매개변수를 stdio_client 함수에 전달하여 읽기 및 쓰기 스트림을 반환하는 컨텍스트를 만듭니다. 이 스트림을 통해 각각 stdio에서 읽고 쓸 수 있습니다.

async def run():
   async with stdio_client(server_params) as (read, write):
       async with ClientSession(read, write) as session:
           # Initialize the connection
           await session.initialize()

           # List available tools
           tools = await session.list_tools()
           print("Available tools:", tools)

이러한 스트림은 ClientSession 클래스로 세션을 만드는 데 사용됩니다. 세션을 만든 후 초기화한 다음 서버의 모든 도구를 나열할 수 있습니다.

이 프로그램을 실행하려면 asyncio 라이브러리를 사용하여 run 함수를 호출하기만 하면 됩니다. 전체 코드는 다음과 같습니다:

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import os

# Create server parameters for stdio connection
server_params = StdioServerParameters(
   command="npx",
   args=["-y", "@novitalabs/novita-mcp-server"],
   env={"NOVITA_API_KEY":  os.environ["NOVITA_API_KEY"]},
)

async def run():
   async with stdio_client(server_params) as (read, write):
       async with ClientSession(read, write) as session:
           # Initialize the connection
           await session.initialize()

           # List available tools
           tools = await session.list_tools()
           print("Available tools:", tools)

if __name__ == "__main__":
   import asyncio

   asyncio.run(run())

스크립트를 실행하면 Novita MCP 서버의 모든 도구를 볼 수 있습니다.

이 예제는 MCP 클라이언트를 사용하는 방법을 보여줍니다. 이 예제에 LLM을 추가하여 효과적으로 호스트로 변환할 수 있지만, 이 튜토리얼의 초점은 MCP 서버를 구축하는 것이므로 다루지 않겠습니다.

FastMCP로 MCP 서버 구축하기

Python MCP SDK에는 MCP 서버를 구축하는 두 가지 방법이 있습니다. 하나는 저수준 서버를 사용하는 방법이고, 다른 하나는 FastMCP를 사용하는 방법입니다. FastMCP는 FastAPI에서 영감을 받아 MCP 서버를 더 쉽게 만들 수 있도록 하는 MCP Python SDK의 클래스입니다.

FastMCP를 사용하여 MCP 서버를 만들어 보겠습니다. 시작하기 전에 MCP 서버의 기능을 고려해 보겠습니다. 현재 Novita MCP 서버는 GPU 관리만 처리하는 도구를 구현합니다. GPU 관리 이상의 작업을 수행하는 Novita API 기반 서버를 구축해 보겠습니다.

대신 Novita 플랫폼의 모델에 대한 액세스를 제공하는 MCP 서버를 구축해 보겠습니다. MCP 서버에는 리소스나 프롬프트 없이 도구만 있습니다. 서버가 가질 도구 목록은 다음과 같습니다:

  • list_models: 이 도구는 Novita 플랫폼의 모든 대규모 언어 모델을 나열합니다.
  • get_model: 이 도구는 특정 대규모 언어 모델을 검색하고 사용합니다.
  • text2image: 이 도구는 주어진 프롬프트에서 이미지를 생성합니다.
  • task_result: 이 도구는 실행 중인 작업의 상태를 ID로 가져오는 데 사용됩니다.
  • text_to_speech: 이 도구는 제공된 텍스트를 음성으로 변환합니다.
  • generate_video: 이 도구는 주어진 프롬프트에서 비디오를 생성합니다.

이제 구축할 도구를 알았으니 서버 생성을 시작해 보겠습니다. 먼저 server.py 파일을 만들고 다음 코드를 추가합니다:

import os
import sys
from mcp.server.fastmcp import FastMCP
import requests
import uvicorn
from starlette.applications import Starlette
from starlette.routing import Mount

base_url = "https://api.novita.ai/v3"

headers = {
   "Content-Type": "application/json",
   "Authorization": f"Bearer {os.environ['NOVITA_API_KEY']}"
}

mcp = FastMCP("Novita_API")

이제 필요한 모듈을 가져오고, API의 기본 URL과 헤더를 정의하고, FastMCP 클래스 인스턴스를 만들었습니다. 도구 생성을 진행해 보겠습니다.

모델 나열

FastMCP에서 도구를 만들려면 FastMCP 인스턴스의 tool 메서드로 함수를 데코레이트해야 합니다. list_models 함수는 모델 나열 엔드포인트를 호출하여 모델 목록을 가져옵니다. 그런 다음 응답을 적절히 형식화하여 호출한 LLM에 전달할 수 있도록 합니다.

@mcp.tool()
def list_models() -> str:
   """
   List all available models from the Novita API.
   """

   url = base_url + "/openai/models"

   response = requests.request("GET", url, headers=headers)
   data = response.json()["data"]
  
   text = ""
   for i, model in enumerate(data, start=1):
       text += f"Model id: {model['id']}\
"
       text += f"Model description: {model['description']}\
"
       text += f"Model type: {model['model_type']}\
\
"
  
   return text

도구의 docstring은 LLM이 도구의 용도를 이해할 수 있도록 기능을 설명하는 역할을 합니다.

모델 가져오기

이 도구는 Novita에 배포된 LLM에 접근하는 데 사용됩니다. 모델 ID와 모델에 대한 프롬프트를 받습니다. Novita 채팅 완료 API 엔드포인트를 사용합니다.

@mcp.tool()
def get_model(model_id: str, message):
   """
   Provide a model ID and a message to get a response from the Novita API.
   """

   url = base_url + "/openai/chat/completions"

   payload = {
       "model": model_id,
       "messages": [
           {
               "content": message,
               "role": "user",
           }
       ],
       "max_tokens": 200,
       "response_format": {
           "type": "text",
       },
   }

   response = requests.request("POST", url, json=payload, headers=headers)
  
   content = response.json()["choices"][0]["message"]["content"]

   return content

Text2Image

이 도구는 프롬프트에서 이미지를 생성합니다. 내부적으로 Novita 텍스트-이미지 엔드포인트를 사용합니다. 이 엔드포인트는 비동기식이므로 이미지를 즉시 반환하지 않고 작업 ID를 반환합니다.

@mcp.tool()
def text2Image(prompt):
   """
   Generate an image from a text prompt using the Novita API.
   """

   url = base_url + "/async/txt2img"

   payload = {
       "request": {
           "model_name": "sd_xl_base_1.0.safetensors",
           "prompt": prompt,
           "width": 1024,
           "height": 1024,
           "image_num": 1,
           "steps": 20,
           "clip_skip": 1,
           "sampler_name": "Euler a",
           "guidance_scale": 7.5,
       },
       "extra": {
           "response_image_type": "jpeg"
       }
   }

   response = requests.request("POST", url, json=payload, headers=headers)

   return response.json()["task_id"]

작업 결과

이 도구는 Novita API에서 진행 중인 작업의 상태를 가져오는 데 사용됩니다. 작업 ID만 받습니다. 예를 들어, Image 도구는 비동기적으로 이미지를 생성하고 호스트에 작업 ID를 반환합니다. 그러면 사용자는 호스트에게 Task Result 도구를 사용하여 해당 작업의 결과를 검색하도록 요청할 수 있습니다.

@mcp.tool()
def task_result(task_id: str):
   """
   Get the current status of a running task using it's task id
   """

   url = base_url + f'/async/task-result?task_id={task_id}'

   response = requests.request("GET", url, headers=headers)
   return response.json()

비디오 생성

이 도구는 Novita API의 Kling AI V1.6 텍스트-비디오 엔드포인트를 사용하여 비디오를 생성합니다. Image 도구와 마찬가지로 비디오를 비동기적으로 생성하고 작업 ID를 반환합니다.

@mcp.tool()
def generateVideo(prompt: str):
   """
   Generate an image using a prompt
   """

   url = base_url + "/async/kling-v1.6-t2v"

   payload = {
       "mode": "Standard",
       "prompt": prompt,
       "negative_prompt": "low quality",
       "guidance_scale": 0.6
   }

   response = requests.post(url, json=payload, headers=headers)
  
   return response.json()

텍스트 음성 변환

이 도구는 Novita API 텍스트-음성 엔드포인트를 사용하여 음성을 생성합니다. 또한 비동기 엔드포인트이므로 작업 ID를 반환합니다.

@mcp.tool()
def textToSpeech(text, voice_id) -> str:
   """
   Generate speech using text and voice id.
   It returns the task id of the generated speech.
  
   The available voice ids are:
   - Emily
   - James
   - Olivia
   - Michael
   - Sarah
   - John
   """

   url = base_url + "/async/txt2speech"

   payload = {
       "request": {
           "voice_id": voice_id,
           "language": "en-US",
           "texts": [text]
       }
   }

   response = requests.post(url, json=payload, headers=headers)
  
   return response.json()["task_id"]

모든 도구를 정의했으면 전송 메커니즘을 설정할 수 있습니다. stdio를 사용하겠습니다.

if __name__ == "__main__":
   # Run using stdio transport
   mcp.run(transport="stdio")

이제 모든 코드를 함께 모을 수 있습니다:

import os
import sys

from mcp.server.fastmcp import FastMCP

import requests

import uvicorn

from starlette.applications import Starlette
from starlette.routing import Mount

base_url = "https://api.novita.ai/v3"

headers = {
   "Content-Type": "application/json",
   "Authorization": f"Bearer {os.environ['NOVITA_API_KEY']}"
}

mcp = FastMCP("Novita_API")

@mcp.tool()
def list_models() -> str:
   """
   List all available models from the Novita API.
   """

   url = base_url + "/openai/models"

   response = requests.request("GET", url, headers=headers)
   data = response.json()["data"]
  
   text = ""
   for i, model in enumerate(data, start=1):
       text += f"Model id: {model['id']}\
"
       text += f"Model description: {model['description']}\
"
       text += f"Model type: {model['model_type']}\
\
"
  
   return text

@mcp.tool()
def get_model(model_id: str, message) -> str:
   """
   Provide a model ID and a message to get a response from the Novita API.
   """

   url = base_url + "/openai/chat/completions"

   payload = {
       "model": model_id,
       "messages": [
           {
               "content": message,
               "role": "user",
           }
       ],
       "max_tokens": 200,
       "response_format": {
           "type": "text",
       },
   }

   response = requests.request("POST", url, json=payload, headers=headers)
  
   content = response.json()["choices"][0]["message"]["content"]

   return content

@mcp.tool()
def text2Image(prompt: str) -> str:
   """
   Generate an image from a text prompt using the Novita API.
   """

   url = base_url + "/async/txt2img"

   payload = {
       "request": {
           "model_name": "sd_xl_base_1.0.safetensors",
           "prompt": prompt,
           "width": 1024,
           "height": 1024,
           "image_num": 1,
           "steps": 20,
           "clip_skip": 1,
           "sampler_name": "Euler a",
           "guidance_scale": 7.5,
       },
       "extra": {
           "response_image_type": "jpeg"
       }
   }

   response = requests.request("POST", url, json=payload, headers=headers)

   return response.json()["task_id"]

@mcp.tool()
def task_result(task_id: str) -> str:
   """
   Get the current status of a running task using it's task id
   """

   url = base_url + f'/async/task-result?task_id={task_id}'

   response = requests.request("GET", url, headers=headers)
   return response.json()

@mcp.tool()
def generateVideo(prompt: str) -> str:
   """
   Generate an image using a prompt
   """

   url = base_url + "/async/kling-v1.6-t2v"

   payload = {
       "mode": "Standard",
       "prompt": prompt,
       "negative_prompt": "low quality",
       "guidance_scale": 0.6
   }

   response = requests.post(url, json=payload, headers=headers)
  
   return response.json()["task_id"]

@mcp.tool()
def textToSpeech(text, voice_id) -> str:
   """
   Generate speech using text and voice id.
   It returns the task id of the generated speech.
  
   The available voice ids are:
   - Emily
   - James
   - Olivia
   - Michael
   - Sarah
   - John
   """

   url = base_url + "/async/txt2speech"

   payload = {
       "request": {
           "voice_id": voice_id,
           "language": "en-US",
           "texts": [text]
       }
   }

   response = requests.post(url, json=payload, headers=headers)
  
   return response.json()["task_id"]

if __name__ == "__main__":
   # Run using stdio transport
   mcp.run(transport="stdio")

다음 설정을 사용하여 Claude Desktop, Cursor 또는 VS Code와 같은 MCP 호스트에서 MCP 서버를 테스트할 수 있습니다:

{
     "command": "python",
     "args": [
       "path/to/server.py" 
     ],
     "env": {
       "NOVITA_API_KEY": "sk_...."
     }
   }

또한 이전에 개발한 클라이언트 스크립트를 사용하여 서버를 테스트할 수도 있습니다.

MCP 저수준 서버 사용하기

이전 섹션에서 구축한 서버는 MCP 서버 구축을 위한 고수준 인터페이스를 제공하는 FastMCP 클래스를 사용합니다. 저수준 서버를 사용하여 MCP 서버를 구축할 수도 있으며, 이를 통해 MCP 프로토콜을 세밀하게 제어할 수 있습니다. 이전 섹션의 서버를 저수준 서버로 수정하는 방법을 살펴보겠습니다.

도구 관리

FastMCP가 수행하는 작업 중 하나는 정의한 도구를 관리하는 것입니다. FastMCP 클래스의 tool 데코레이터로 함수를 장식하면 FastMCP가 해당 도구를 목록에 추가합니다. 에이전트가 해당 도구를 요청하면 FastMCP가 도구를 가져와 호출한 다음 결과를 다시 에이전트로 보냅니다.

저수준 서버에서는 call_tool 데코레이터를 사용하여 이 프로세스를 직접 관리할 수 있습니다.

@app.call_tool()
async def manage_tool(name: str, arguments: dict ) -> list[types.TextContent]:
   if name == "list_models":
       return await list_models_tool()
  
   if name == "get_model":
       return await get_model_tool(arguments)
  
   else:
       raise ValueError(f"Unknown tool: {name}")

위 코드는 call_tool 메서드로 데코레이트된 함수를 보여줍니다. 에이전트가 도구를 호출하면 도구 이름과 도구가 예상하는 인수를 전달합니다. 함수 이름을 사용하여 에이전트가 호출하려는 도구를 확인하고 실행할 수 있습니다.

또한 저수준 서버를 사용하면 FastMCP와 달리 도구 나열도 직접 관리할 수 있습니다. FastMCP는 tool 메서드로 함수가 데코레이트되면 자동으로 수행합니다.

@app.list_tools()
async def list_tools() -> list[types.Tool]:
   return [
       types.Tool(
           name="list_models",
           description="List all available models from the Novita API.",
           inputSchema={"type": "object", "properties": {}},
       ),
       types.Tool(
           name="get_model",
           description="Provide a model ID and a message to get a response from the Novita API.",
           inputSchema={
               "type": "object",
               "required": ["model_id", "message"],
               "properties": {
                   "model_id": {
                       "type": "string",
                       "description": "The ID of the model to use.",
                   },
                   "message": {
                       "type": "string",
                       "description": "The input message to send.",
                   },
               },
           },
       ),
   ]

전체 저수준 서버 코드는 다음과 같습니다. 두 개의 도구가 포함되어 있으며 표준 입력 및 출력을 통해 클라이언트와 통신할 수 있습니다.

import os
import asyncio
import requests
from mcp.server.lowlevel import Server
from mcp.server.stdio import stdio_server
import mcp.types as types

base_url = "https://api.novita.ai/v3"

headers = {
   "Content-Type": "application/json",
   "Authorization": f"Bearer {os.environ['NOVITA_API_KEY']}"
}

app = Server("Novita_API")

async def list_models_tool():
   """
   Lists all available models from the Novita API.
   """
   url = base_url + "/openai/models"
   response = requests.get(url, headers=headers)
   data = response.json()["data"]

   text = ""
   for i, model in enumerate(data, start=1):
       text += f"Model id: {model['id']}\
"
       text += f"Model description: {model['description']}\
"
       text += f"Model type: {model['model_type']}\
\
"

   return [types.TextContent(type="text", text=text)]

async def get_model_tool(arguments: dict):
   """
   Given a model ID and a user message, fetch a response from the Novita API.
   """
   model_id = arguments.get("model_id")
   message = arguments.get("message")

   if not model_id or not message:
       raise ValueError("Both 'model_id' and 'message' are required.")

   url = base_url + "/openai/chat/completions"

   payload = {
       "model": model_id,
       "messages": [
           {
               "content": message,
               "role": "user",
           }
       ],
       "max_tokens": 200,
       "response_format": {
           "type": "text",
       },
   }

   response = requests.post(url, json=payload, headers=headers)
   content = response.json()["choices"][0]["message"]["content"]
   return [types.TextContent(type="text", text=content)]

@app.call_tool()
async def manage_tool(name: str, arguments: dict ) -> list[types.TextContent]:
   if name == "list_models":
       return await list_models_tool()
  
   if name == "get_model":
       return await get_model_tool(arguments)
  
   else:
       raise ValueError(f"Unknown tool: {name}")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
   return [
       types.Tool(
           name="list_models",
           description="List all available models from the Novita API.",
           inputSchema={"type": "object", "properties": {}},
       ),
       types.Tool(
           name="get_model",
           description="Provide a model ID and a message to get a response from the Novita API.",
           inputSchema={
               "type": "object",
               "required": ["model_id", "message"],
               "properties": {
                   "model_id": {
                       "type": "string",
                       "description": "The ID of the model to use.",
                   },
                   "message": {
                       "type": "string",
                       "description": "The input message to send.",
                   },
               },
           },
       ),
   ]

async def main():
   async with stdio_server() as streams:
       await app.run(streams[0], streams[1], app.create_initialization_options())

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

저수준 서버는 도구 관리에 대한 세밀한 제어를 제공할 뿐만 아니라 리소스, 프롬프트 및 수명 주기 관리와 같은 MCP 프로토콜의 다른 부분에 대한 저수준 제어도 제공합니다.

결론

이 튜토리얼에서는 Novita API를 사용하여 모든 MCP 호스트의 기능을 향상시키는 MCP 서버를 구축했습니다. 도구를 정의하고 노출하는 방법을 포함하여 MCP 서버를 만드는 데 필요한 기본 사항을 배웠습니다.

이 기초를 바탕으로 인증, 원격 MCP 서버 배포, 저수준 Python 구현 직접 작업과 같은 고급 주제를 탐색할 수 있습니다.

또한 Novita API 자체를 살펴보고 다양한 언어 모델 세트부터 비디오, 오디오 및 이미지 생성 도구에 이르기까지 그 기능을 확인했습니다. 이미지 및 얼굴 편집기 엔드포인트와 같이 다루지 않은 다른 API를 사용해 보려면 Novita LLM 플레이그라운드를 방문하세요.

Novita AI는 간단한 API를 통해 개발자가 AI 모델을 쉽게 배포할 수 있는 AI 클라우드 플랫폼이며, 구축 및 확장을 위한 저렴하고 안정적인 GPU 클라우드를 제공합니다.