模型上下文協定（Model Context Protocol，MCP）正迅速成為 AI 領域中最受矚目的概念之一。儘管它相對新穎，卻已吸引開發者與各大科技公司的高度關注。
MCP 由 Anthropic 創建，是一種開放標準協定，類似 HTTP 等其他協定。然而，HTTP 是設計來經由伺服器和網頁瀏覽器（或 HTTP 用戶端）將使用者連接到資源，而 MCP 則是設計來將大型語言模型（LLMs）連接到外部工具和資料。
假設你有一個聊天機器人，並希望它能存取 GitHub。你希望使用者能透過聊天機器人建立議題、關閉議題、新增註解，或檢視現有的拉取請求。為了讓聊天機器人能執行所有這些操作，它需要能存取相關工具。
這就是 MCP 伺服器的用武之地。
MCP 伺服器託管多個工具，聊天機器人可以利用其內部的 LLM 來呼叫這些工具，從而獲得上述能力。以 GitHub 為例，我們可以使用官方的 GitHub MCP 伺服器來執行所有前述任務。
在本文中，我們將使用 Novita API 來建置一個 MCP 伺服器。透過這個 MCP 伺服器，任何支援 MCP 協定的應用程式都能存取 Novita 上所有可用的 LLM、生成圖片、建立影片以及執行語音合成。
了解 MCP 架構
MCP 架構 [來源]
MCP 架構類似於典型的客戶端-伺服器設定，包含四個重要元件：
- MCP 用戶端 (MCP Client)
- MCP 伺服器 (MCP Server)
- MCP 傳輸層 (MCP Transports)
- MCP 主機 (MCP Host)
MCP 用戶端
MCP 用戶端是 AI 應用程式與 MCP 伺服器之間的通訊橋樑。一旦建立通訊，應用程式就能存取伺服器所提供的所有工具和資源。
MCP 伺服器
MCP 伺服器託管所有用戶端可提供給 AI 應用程式的工具和資料，包含以下項目：
- Tools (工具)
- Resources (資源)
- Prompts (提示)
工具
工具為 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 傳輸 [來源]
MCP 用戶端和伺服器可以透過標準輸入和輸出 (STDIO) 進行通訊。使用 STDIO 時，用戶端和伺服器都在同一台機器上執行。用戶端將所有請求寫入 stdin，而伺服器則將回應寫入 stdout。
Streamable HTTP
使用 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 中。
# 建立 stdio 連線的伺服器參數
server_params = StdioServerParameters(
command="npx",
args=["-y", "@novitalabs/novita-mcp-server"],
env={"NOVITA_API_KEY": os.environ["NOVITA_API_KEY"]},
)
接下來，建立一個非同步函式。在函式中，將伺服器參數傳遞給 stdio_client 函式，該函式會建立一個 context，回傳讀取和寫入串流。這些串流讓我們能分別從 stdio 讀取和寫入。
async def run():
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# 初始化連線
await session.initialize()
# 列出可用的工具
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
# 建立 stdio 連線的伺服器參數
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:
# 初始化連線
await session.initialize()
# 列出可用的工具
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 是 MCP Python SDK 中的一個類別，靈感來自 FastAPI，目的是讓建立 MCP 伺服器更容易。
讓我們使用 FastMCP 來建立我們的 MCP 伺服器。在開始之前，先考慮一下 MCP 伺服器的功能。目前的 Novita MCP 伺服器只實作了處理 GPU 管理的工具。我們試著圍繞 Novita API 建立一個超越 GPU 管理的伺服器。
取而代之的是，我們建立一個能存取 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:
"""
列出 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
工具中的文件字串是用來描述工具功能的方式，讓 LLM 能了解該工具的用途。
取得模型
此工具用於存取 Novita 上已部署的 LLM。它接收模型 ID 和提供給模型的提示。它使用 Novita 聊天完成 API 端點。
@mcp.tool()
def get_model(model_id: str, message):
"""
提供模型 ID 和訊息，從 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):
"""
使用 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 給主機。使用者可以要求主機使用任務結果工具來擷取該任務的結果。
@mcp.tool()
def task_result(task_id: str):
"""
使用任務 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):
"""
使用提示生成圖片。
"""
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:
"""
使用文字和語音 ID 生成語音。
它會回傳所生成語音的任務 ID。
可用的語音 ID 有：
- 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__":
# 使用 stdio 傳輸執行
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:
"""
列出 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:
"""
提供模型 ID 和訊息，從 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:
"""
使用 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:
"""
使用任務 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:
"""
使用提示生成圖片。
"""
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:
"""
使用文字和語音 ID 生成語音。
它會回傳所生成語音的任務 ID。
可用的語音 ID 有：
- 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__":
# 使用 stdio 傳輸執行
mcp.run(transport="stdio")
我們可以在任何 MCP 主機上測試我們的 MCP 伺服器，例如 Claude Desktop、Cursor 或 VS Code，使用以下設定：
{
"command": "python",
"args": [
"path/to/server.py"
],
"env": {
"NOVITA_API_KEY": "sk_...."
}
}
我們也可以使用之前開發的用戶端腳本來測試我們的伺服器。
使用 MCP 低階伺服器
我們在上一節中建立的伺服器使用了 FastMCP 類別，它提供了建置 MCP 伺服器的高階介面。你也可以使用低階伺服器來建置 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 不同（當函式使用 tool 方法裝飾時，FastMCP 會自動處理），使用低階伺服器時，我們還需要管理工具的列出。
@app.list_tools()
async def list_tools() -> list[types.Tool]:
return [
types.Tool(
name="list_models",
description="列出 Novita API 所有可用的模型。",
inputSchema={"type": "object", "properties": {}},
),
types.Tool(
name="get_model",
description="提供模型 ID 和訊息，從 Novita API 取得回應。",
inputSchema={
"type": "object",
"required": ["model_id", "message"],
"properties": {
"model_id": {
"type": "string",
"description": "要使用的模型 ID。",
},
"message": {
"type": "string",
"description": "要傳送的輸入訊息。",
},
},
},
),
]
以下是完整的低階伺服器程式碼。它包含兩個工具，並能透過標準輸入和輸出與用戶端通訊。
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():
"""
列出 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):
"""
給定模型 ID 和使用者訊息，從 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="列出 Novita API 所有可用的模型。",
inputSchema={"type": "object", "properties": {}},
),
types.Tool(
name="get_model",
description="提供模型 ID 和訊息，從 Novita API 取得回應。",
inputSchema={
"type": "object",
"required": ["model_id", "message"],
"properties": {
"model_id": {
"type": "string",
"description": "要使用的模型 ID。",
},
"message": {
"type": "string",
"description": "要傳送的輸入訊息。",
},
},
},
),
]
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 協定其他部分（如生命週期管理）的低階控制。
結論
在本教學中，我們建立了一個 MCP 伺服器，它使用 Novita API 來增強任何 MCP 主機的能力。你已經學到了建立 MCP 伺服器的基本知識，包括如何定義和公開工具。
有了這個基礎，你可以開始探索更高階的主題，例如認證、部署遠端 MCP 伺服器，以及直接使用低階 Python 實作。
我們也探索了 Novita API 本身，並看到了它的能力，從多樣化的語言模型到影片、音訊和圖片的生成工具。造訪 Novita LLM Playground 來試試我們沒有涵蓋到的其他 API，例如圖片和臉部編輯端點。
Novita AI 是一個 AI 雲端平台，為開發者提供透過簡單 API 部署 AI 模型的簡易方式，同時也提供經濟實惠且可靠的 GPU 雲端來建置和擴展。
