模型上下文协议（Model Context Protocol，简称 MCP）正在迅速成为人工智能领域最受关注的概念之一。尽管它相对较新，但已经吸引了众多开发者以及大型科技公司的关注。
MCP 由 Anthropic 创建，是一个开放标准协议，类似于 HTTP 等其他协议。不过，HTTP 旨在通过服务器和 Web 浏览器（或 HTTP 客户端）将用户连接到资源，而 MCP 则旨在将大型语言模型（LLMs）连接到外部工具和数据。
假设你有一个聊天机器人，希望它能够访问 GitHub。你希望用户能够使用该聊天机器人来创建问题、关闭问题、添加评论或查看现有的拉取请求。要让聊天机器人执行所有这些操作，它需要获得对工具的访问权限。
这时就需要用到 MCP 服务器了。
MCP 服务器托管了多个工具，聊天机器人可以通过其内部 LLM 调用这些工具，从而获得这些能力。以 GitHub 为例，我们可以使用官方的 GitHub MCP 服务器来执行上述所有任务。
在本文中，我们将使用 Novita API 构建一个 MCP 服务器。借助这个 MCP 服务器，任何支持 MCP 协议的应用程序都将能够访问 Novita 上所有可用的 LLM、生成图像、创建视频以及进行语音合成。
理解 MCP 架构
MCP 架构 [来源]
MCP 架构类似于典型的客户端-服务器设置，包含四个重要组件：
- MCP Client（MCP 客户端）
- MCP Server（MCP 服务器）
- MCP Transports（MCP 传输）
- MCP Host（MCP 主机）
MCP 客户端
MCP 客户端充当 AI 应用程序与 MCP 服务器之间的通信网关。一旦通信建立，应用程序即可访问服务器拥有的所有工具和资源。
MCP 服务器
MCP 服务器托管客户端可以提供给 AI 应用程序的所有工具和数据。它包含以下内容：
- Tools（工具）
- Resources（资源）
- Prompts（提示）
工具
工具为 LLM 提供了原本不具备的外部能力，例如获取时间、更新和读取数据库、查询天气等。这些工具本质上是程序员定义的、可供调用的函数。
一个 MCP 服务器可以托管多个工具。这些工具使得 LLM 能够像代理（agent）一样工作，因为它可以执行各种操作。
工具由代理管理，使用过程中可能涉及可选的人工监督。它们类似于 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 传输 [来源]
MCP 客户端和服务器可以通过标准输入和输出（STDIO）进行通信。使用 STDIO 时，客户端和服务器都运行在同一台机器上。客户端将所有请求写入标准输入（stdin），而服务器则将响应写入标准输出（stdout）。
Streamable HTTP
使用 Streamable HTTP 的 MCP 传输 [来源]
Streamable HTTP 允许 MCP 客户端和服务器通过 HTTP 进行通信。它使用 HTTP POST 方法从客户端发送请求并接收服务器响应。必要时，它可以选择切换到服务器发送事件（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 函数，该函数将创建一个返回读取和写入流的上下文。这些流使我们能够分别从 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("可用的工具:", 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("可用的工具:", 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 管理的服务器。
相反，让我们构建一个 MCP 服务器，提供对 Novita 平台上模型的访问。我们的 MCP 服务器将只包含工具，没有资源或提示。以下是服务器将拥有的工具列表：
- list_models：此工具将列出 Novita 平台上的所有大型语言模型。
- get_model：此工具将检索并使用特定的 LLM。
- 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
工具中的文档字符串用于描述工具的功能，以便 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
文生图
此工具根据提示生成图像。它在底层使用 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。例如，图像工具异步生成图像并向主机返回一个任务 ID。用户可以要求主机使用任务结果工具检索该任务的结果。
@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 文生视频端点生成视频。与图像工具一样，它异步生成视频并返回一个任务 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__":
# 使用 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:
"""
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__":
# 使用 stdio 传输运行
mcp.run(transport="stdio")
我们可以在任何 MCP 主机（如 Claude Desktop、Cursor 或 VS Code）上测试我们的 MCP 服务器，使用以下配置：
{
"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 不同，使用低层服务器时，我们还需要自行管理工具的列表，而 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 协议其他部分（如生命周期管理）的低层控制。
结论
在本教程中，我们构建了一个 MCP 服务器，它使用 Novita API 增强了任何 MCP 主机的功能。你已经了解了创建 MCP 服务器所需的基础知识，包括如何定义和暴露工具。
有了这个基础，你可以开始探索更高级的主题，例如身份验证、部署远程 MCP 服务器，以及直接使用低层 Python 实现。
我们还探索了 Novita API 本身，看到了它的能力，从多样化的语言模型到视频、音频和图像的生成工具。欢迎访问 Novita LLM Playground 来尝试我们尚未涉及的其他 API，例如图像和面部编辑器端点。
Novita AI 是一个 AI 云平台，为开发者提供通过简单 API 部署 AI 模型的简易方式，同时还提供经济实惠且可靠的 GPU 云用于构建自定义模型。
