MCP 模型上下文协议：AI 原生开发的技术革命与实践

// 引入核心依赖

import { FastMcp } from "fast-mcp"; // MCP服务器构建库

import { z } from "zod"; // 参数校验库

// 初始化MCP服务器实例

const server = new FastMcp({

name: "calculator-server", // 服务名称

version: "1.0.0", // 版本号

description: "基础加减运算MCP服务"

});

// 注册加法工具

server.addTool({

name: "addition", // 工具唯一标识

description: "计算两个数字的和，优先处理加法问题",

parameters: z.object({ // 参数校验schema

a: z.number().describe("被加数，如3.14"),

b: z.number().describe("加数，如2.86")

}),

execute: async (args) => { // 执行逻辑

return `计算结果：${args.a + args.b}`;

}

});

// 注册减法工具

server.addTool({

name: "subtraction",

description: "计算两个数字的差，优先处理减法问题",

parameters: z.object({

a: z.number().describe("被减数"),

b: z.number().describe("减数")

}),

execute: async (args) => {

return `计算结果：${args.a - args.b}`;

}

});

// 启动服务（stdio传输模式适配本地客户端）

server.start({

transportType: "stdio",

port: 3000 // 可选，默认随机端口

});

"calculator": {

"command": "node",

"args": ["D:/mcp-server/calculator-server.js"]

}

# 基于Python的多模态MCP服务器

from fastapi import FastAPI, HTTPException

from pydantic import BaseModel

import openai

import os

from datetime import datetime

import json

# 初始化FastAPI（MCP服务器基础框架）

app = FastAPI(title="AI-Image-Gen MCP Server", version="0.1.0")

# 配置OpenAI API密钥

openai.api_key = os.getenv("OPENAI_API_KEY")

# 定义请求模型（MCP工具参数规范）

class ImageRequest(BaseModel):

prompt: str

model: str = "dall-e-3" # 默认使用DALL·E 3

size: str = "1024x1024" # 默认分辨率

style: str = "vivid" # 仅DALL·E 3支持

n: int = 1 # 生成数量

# 定义存储配置

STORAGE_DIR = "./generated_images"

os.makedirs(STORAGE_DIR, exist_ok=True)

# 注册MCP工具端点（工具发现接口）

@app.get("/mcp/resources")

async def list_resources():

"""暴露MCP资源清单（工具元数据）"""

return {

"tools": [

{

"name": "generate_image",

"description": "多模型图像生成工具，支持DALL·E 3/2与GPT-Image-1",

"parameters": {

"type": "object",

"properties": {

"prompt": {"type": "string", "description": "图像描述提示词"},

"model": {"type": "string", "enum": ["dall-e-3", "dall-e-2", "gpt-image-1"]},

"size": {"type": "string", "enum": ["256x256", "512x512", "1024x1024", "1792x1024"]},

"style": {"type": "string", "enum": ["vivid", "natural"]},

"n": {"type": "integer", "minimum": 1, "maximum": 10}

},

"required": ["prompt"]

}

}

]

}

# 注册工具执行端点

@app.post("/mcp/tools/generate_image/execute")

async def generate_image(request: ImageRequest):

"""执行图像生成任务"""

try:

# 模型路由逻辑

if request.model == "dall-e-3":

response = openai.images.generate(

model=request.model,

prompt=request.prompt,

size=request.size,

style=request.style,

n=request.n,

response_format="url"

)

elif request.model == "dall-e-2":

response = openai.images.generate(

model=request.model,

prompt=request.prompt,

size=request.size,

n=request.n,

response_format="url"

)

elif request.model == "gpt-image-1":

response = openai.chat.completions.create(

model=request.model,

messages=[{"role": "user", "content": f"生成图像: {request.prompt}"}],

response_format={"type": "image_url"}

)

else:

raise HTTPException(status_code=400, detail="不支持的模型类型")

# 处理生成结果

timestamp = datetime.now().strftime("%Y%m%d%H%M%S")

results = []

for i, item in enumerate(response.data):

# 存储元数据

meta = {

"prompt": request.prompt,

"model": request.model,

"size": request.size,

"timestamp": timestamp,

"image_url": item.url

}

meta_path = os.path.join(STORAGE_DIR, f"image_{timestamp}_{i}_meta.json")

with open(meta_path, "w") as f:

json.dump(meta, f)

results.append(meta)

return {"status": "success", "results": results}

except Exception as e:

return {"status": "error", "message": str(e)}

# MCP服务器启动入口

if __name__ == "__main__":

import uvicorn

uvicorn.run(app, host="0.0.0.0", port=8000)

# MCP任务编排服务器核心模块

import sqlite3

import yaml

from typing import Dict, List, Any

from pydantic import BaseModel

# 1. 角色定义模块（可通过YAML配置扩展）

class AIRole(BaseModel):

name: str

system_prompt: str

capabilities: List[str]

def load_roles(role_path: str = ".task_orchestrator/roles/project_roles.yaml") -> Dict[str, AIRole]:

"""加载AI角色配置"""

with open(role_path, "r") as f:

roles_data = yaml.safe_load(f)

return {r["name"]: AIRole(**r) for r in roles_data["roles"]}

# 2. 任务分解模块

class TaskDecomposer:

def __init__(self, llm_client):

self.llm_client = llm_client

self.roles = load_roles()

def decompose(self, user_query: str) -> List[Dict[str, Any]]:

"""将复杂任务分解为角色化子任务"""

decomposition_prompt = f"""

将用户需求分解为结构化工作流，分配给对应AI角色：

用户需求：{user_query}

可用角色：{[r for r in self.roles.keys()]}

输出格式：[{"步骤序号": "角色名", "任务描述": "...", "输出物": "..."}]

"""

response = self.llm_client.chat.completions.create(

model="claude-3-opus",

messages=[{"role": "user", "content": decomposition_prompt}]

)

return eval(response.choices[0].message.content)

# 3. 任务执行模块

class TaskExecutor:

def __init__(self, llm_client, db_path: str = "task_orchestrator.db"):

self.llm_client = llm_client

self.roles = load_roles()

self.db_conn = sqlite3.connect(db_path)

self._init_db()

def _init_db(self):

"""初始化任务数据库"""

cursor = self.db_conn.cursor()

cursor.execute("""

CREATE TABLE IF NOT EXISTS tasks (

task_id TEXT PRIMARY KEY,

user_query TEXT,

decomposition TEXT,

status TEXT,

created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP

)

""")

cursor.execute("""

CREATE TABLE IF NOT EXISTS artifacts (

artifact_id TEXT PRIMARY KEY,

task_id TEXT,

role_name TEXT,

content TEXT,

file_path TEXT,

created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,

FOREIGN KEY (task_id) REFERENCES tasks(task_id)

)

""")

self.db_conn.commit()

def execute_subtask(self, task_id: str, subtask: Dict[str, Any]) -> str:

"""执行单个子任务并存储结果"""

role = self.roles[subtask["角色名"]]

response = self.llm_client.chat.completions.create(

model="claude-3-sonnet",

messages=[

{"role": "system", "content": role.system_prompt},

{"role": "user", "content": subtask["任务描述"]}

]

)

content = response.choices[0].message.content

# 存储输出物

artifact_id = f"art_{task_id}_{subtask['步骤序号']}"

cursor = self.db_conn.cursor()

cursor.execute("""

INSERT INTO artifacts (artifact_id, task_id, role_name, content)

VALUES (?, ?, ?, ?)

""", (artifact_id, task_id, subtask["角色名"], content))

self.db_conn.commit()

return content

# 4. MCP服务封装

from fastapi import FastAPI

app = FastAPI(title="MCP Task Orchestrator", version="1.0.0")

llm_client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

decomposer = TaskDecomposer(llm_client)

executor = TaskExecutor(llm_client)

@app.post("/mcp/tools/orchestrate_task/execute")

async def orchestrate_task(user_query: str):

"""MCP任务编排工具执行接口"""

task_id = f"task_{datetime.now().strftime('%Y%m%d%H%M%S')}"

# 1. 任务分解

subtasks = decomposer.decompose(user_query)

# 2. 存储任务信息

cursor = executor.db_conn.cursor()

cursor.execute("""

INSERT INTO tasks (task_id, user_query, decomposition, status)

VALUES (?, ?, ?, ?)

""", (task_id, user_query, str(subtasks), "in_progress"))

executor.db_conn.commit()

# 3. 执行子任务

results = []

for subtask in subtasks:

result = executor.execute_subtask(task_id, subtask)

results.append({

"步骤": subtask["步骤序号"],

"角色": subtask["角色名"],

"结果": result[:50] + "..." if len(result) > 50 else result

})

# 4. 更新任务状态

cursor.execute("""

UPDATE tasks SET status = ? WHERE task_id = ?

""", ("completed", task_id))

executor.db_conn.commit()

return {

"task_id": task_id,

"subtasks": results,

"status": "completed"

}

# MCP服务认证中间件

from fastapi import Depends, HTTPException

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_mcp_client(token: str = Depends(oauth2_scheme)):

"""验证MCP客户端身份"""

if not token.startswith("MCP-SECRET-"):

raise HTTPException(status_code=401, detail="无效的MCP客户端令牌")

# 进一步校验令牌有效性（如查询数据库）

return token

# 在工具接口中应用

@app.post("/mcp/tools/secure_tool/execute")

async def secure_tool(token: str = Depends(verify_mcp_client)):

pass

// 防止命令注入的参数清洗

function sanitizeParameters(params) {

const dangerousChars = /[;|`$&*()<>[]{}'"]/g;

return Object.fromEntries(

Object.entries(params).map(([k, v]) =>

[k, typeof v === "string" ? v.replace(dangerousChars, "") : v]

)

);

}

// 执行前调用清洗

execute: async (args) => {

const safeArgs = sanitizeParameters(args);

return await runCommand(safeArgs.command);

}