49f922444a
- 商品上架 MCP: 简化 batchPublish, price=0 默认 0.01, sku 用 spec/name - 集成 service: 完善 import_products 调用链 - 前端 ProductPage: 调整按钮与列表 - 文档: 本地开发工作流 + devbox F002 升级需求 - 脚本: rotate-mcp-key.sh (一键密钥轮换)
454 lines
13 KiB
TypeScript
454 lines
13 KiB
TypeScript
import { Injectable, Logger, NotFoundException, OnModuleDestroy } from '@nestjs/common'
|
||
import { PrismaService } from '../prisma/prisma.service'
|
||
import { Client } from '@modelcontextprotocol/sdk/client/index.js'
|
||
import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js'
|
||
import { createCipheriv, createDecipheriv, randomBytes } from 'crypto'
|
||
|
||
interface MCPClientInstance {
|
||
client: Client
|
||
transport: any
|
||
connectedAt: Date
|
||
}
|
||
|
||
@Injectable()
|
||
export class IntegrationService implements OnModuleDestroy {
|
||
private readonly logger = new Logger(IntegrationService.name)
|
||
private mcpClients: Map<string, MCPClientInstance> = new Map()
|
||
private readonly encryptKey: Buffer
|
||
|
||
constructor(private prisma: PrismaService) {
|
||
const key = process.env.ENCRYPT_KEY
|
||
if (!key || key.length < 64) {
|
||
throw new Error('ENCRYPT_KEY 环境变量缺失或长度不足 (需要 64 hex = 32 bytes)')
|
||
}
|
||
this.encryptKey = Buffer.from(key, 'hex')
|
||
}
|
||
|
||
private encrypt(plaintext: string): string {
|
||
const iv = randomBytes(12)
|
||
const cipher = createCipheriv('aes-256-gcm', this.encryptKey, iv)
|
||
const encrypted = Buffer.concat([cipher.update(plaintext, 'utf8'), cipher.final()])
|
||
const tag = cipher.getAuthTag()
|
||
return Buffer.concat([iv, tag, encrypted]).toString('base64')
|
||
}
|
||
|
||
private decrypt(encoded: string): string {
|
||
const buf = Buffer.from(encoded, 'base64')
|
||
const iv = buf.subarray(0, 12)
|
||
const tag = buf.subarray(12, 28)
|
||
const ciphertext = buf.subarray(28)
|
||
const decipher = createDecipheriv('aes-256-gcm', this.encryptKey, iv)
|
||
decipher.setAuthTag(tag)
|
||
return decipher.update(ciphertext) + decipher.final('utf8')
|
||
}
|
||
|
||
// 加密/解密 appSecret (对应 DB 字段 apiKey, 名字沿用历史)
|
||
private encryptApiKey(apiKey?: string | null): string | null {
|
||
if (!apiKey) return null
|
||
if (apiKey.length > 40 && /^[A-Za-z0-9+/=]+$/.test(apiKey)) return apiKey
|
||
return this.encrypt(apiKey)
|
||
}
|
||
|
||
private decryptApiKey(apiKey?: string | null): string | null {
|
||
if (!apiKey) return null
|
||
try {
|
||
return this.decrypt(apiKey)
|
||
} catch {
|
||
return apiKey
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 构造 MCP transport 时使用的认证 header
|
||
* 多租户 APIKey: 用 x-app-id + x-app-secret 替代旧的 x-api-key
|
||
* x-customer-code 不再传(由 MCP 验证后从 DB 取并以 X-Customer-Code 头传给 Java 后端)
|
||
*/
|
||
private buildAuthHeaders(integration: { appId?: string | null; apiKey?: string | null; customerCode?: string | null }) {
|
||
const headers: Record<string, string> = {}
|
||
if (integration.appId) headers['x-app-id'] = integration.appId
|
||
if (integration.apiKey) {
|
||
const secret = this.decryptApiKey(integration.apiKey)
|
||
if (secret) headers['x-app-secret'] = secret
|
||
}
|
||
return headers
|
||
}
|
||
|
||
async onModuleDestroy() {
|
||
for (const [id, instance] of this.mcpClients) {
|
||
try {
|
||
await instance.client.close()
|
||
} catch (e) {
|
||
this.logger.warn(`关闭 MCP 连接 ${id} 失败: ${e.message}`)
|
||
}
|
||
}
|
||
this.mcpClients.clear()
|
||
}
|
||
|
||
async create(data: {
|
||
name: string
|
||
type?: string
|
||
endpoint: string
|
||
appId?: string
|
||
apiKey?: string // appSecret
|
||
customerCode?: string
|
||
}) {
|
||
return this.prisma.integration.create({
|
||
data: {
|
||
name: data.name,
|
||
type: data.type || 'mcp',
|
||
endpoint: data.endpoint,
|
||
appId: data.appId,
|
||
apiKey: this.encryptApiKey(data.apiKey),
|
||
customerCode: data.customerCode,
|
||
status: 'inactive',
|
||
},
|
||
})
|
||
}
|
||
|
||
async findAll() {
|
||
const rows = await this.prisma.integration.findMany({
|
||
orderBy: { createdAt: 'desc' },
|
||
})
|
||
return rows.map(r => ({ ...r, apiKey: this.decryptApiKey(r.apiKey) }))
|
||
}
|
||
|
||
async findOne(id: string) {
|
||
const integration = await this.prisma.integration.findUnique({
|
||
where: { id },
|
||
})
|
||
if (!integration) throw new NotFoundException('对接配置不存在')
|
||
return { ...integration, apiKey: this.decryptApiKey(integration.apiKey) }
|
||
}
|
||
|
||
async update(id: string, data: {
|
||
name?: string
|
||
endpoint?: string
|
||
appId?: string
|
||
apiKey?: string
|
||
customerCode?: string
|
||
}) {
|
||
await this.findOne(id)
|
||
return this.prisma.integration.update({
|
||
where: { id },
|
||
data: {
|
||
...data,
|
||
apiKey: data.apiKey !== undefined ? this.encryptApiKey(data.apiKey) : undefined,
|
||
},
|
||
})
|
||
}
|
||
|
||
async remove(id: string) {
|
||
await this.findOne(id)
|
||
await this.disconnect(id)
|
||
return this.prisma.integration.delete({
|
||
where: { id },
|
||
})
|
||
}
|
||
|
||
/**
|
||
* 连接 MCP server (多租户 APIKey 模式)
|
||
*/
|
||
async connect(id: string): Promise<{ success: boolean; tools?: any[]; error?: string }> {
|
||
const integration = await this.findOne(id)
|
||
|
||
if (!integration.appId || !integration.apiKey) {
|
||
const msg = '缺少 appId 或 appSecret,请先在 devbox B2B 运营后台"集成管理"申请 APIKey 后填入'
|
||
await this.prisma.integration.update({
|
||
where: { id },
|
||
data: { status: 'error', error: msg },
|
||
})
|
||
return { success: false, error: msg }
|
||
}
|
||
|
||
try {
|
||
await this.disconnect(id)
|
||
|
||
const client = new Client(
|
||
{ name: 'platform-assistant', version: '1.0.0' },
|
||
{ capabilities: {} }
|
||
)
|
||
|
||
const transport = new StreamableHTTPClientTransport(
|
||
new URL(integration.endpoint),
|
||
{
|
||
requestInit: {
|
||
headers: this.buildAuthHeaders(integration),
|
||
},
|
||
}
|
||
)
|
||
|
||
await client.connect(transport)
|
||
|
||
const toolsResult = await client.listTools()
|
||
const tools = toolsResult.tools || []
|
||
|
||
// Persist companyUserId locally so lead publishToDirev can auto-fill enterpriseId/userId
|
||
let companyUserId: string | null = null
|
||
let customerCode: string | null = null
|
||
try {
|
||
const identityResult: any = await client.callTool({
|
||
name: 'get_app_identity',
|
||
arguments: {},
|
||
})
|
||
const identityText = identityResult?.content?.[0]?.text
|
||
if (identityText) {
|
||
const identity = JSON.parse(identityText)
|
||
companyUserId = identity.companyUserId ? String(identity.companyUserId) : null
|
||
customerCode = identity.customerCode || null
|
||
}
|
||
} catch (e: any) {
|
||
this.logger.warn(`get_app_identity 调用失败 (非致命): ${e.message}`)
|
||
}
|
||
|
||
this.mcpClients.set(id, {
|
||
client,
|
||
transport,
|
||
connectedAt: new Date(),
|
||
})
|
||
|
||
await this.prisma.integration.update({
|
||
where: { id },
|
||
data: {
|
||
status: 'active',
|
||
tools: JSON.stringify(tools),
|
||
lastConnected: new Date(),
|
||
error: null,
|
||
companyUserId,
|
||
customerCode: customerCode || integration.customerCode,
|
||
},
|
||
})
|
||
|
||
this.logger.log(`MCP 连接成功: ${integration.name} (appId=${integration.appId}), ${tools.length} 个工具, companyUserId=${companyUserId || 'null'}`)
|
||
|
||
return { success: true, tools }
|
||
} catch (e: any) {
|
||
this.logger.error(`MCP 连接失败: ${integration.name}`, e.stack)
|
||
await this.prisma.integration.update({
|
||
where: { id },
|
||
data: { status: 'error', error: e.message },
|
||
})
|
||
return { success: false, error: e.message }
|
||
}
|
||
}
|
||
|
||
async disconnect(id: string) {
|
||
const instance = this.mcpClients.get(id)
|
||
if (instance) {
|
||
try {
|
||
await instance.client.close()
|
||
} catch (e) {
|
||
this.logger.warn(`关闭 MCP 连接 ${id} 失败: ${e.message}`)
|
||
}
|
||
this.mcpClients.delete(id)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 测试连接 (不保存状态) - 多租户 APIKey
|
||
*/
|
||
async testConnection(
|
||
endpoint: string,
|
||
appId: string,
|
||
appSecret: string
|
||
): Promise<{ success: boolean; tools?: any[]; error?: string }> {
|
||
let client: Client | null = null
|
||
let transport: any = null
|
||
|
||
try {
|
||
client = new Client(
|
||
{ name: 'platform-assistant-test', version: '1.0.0' },
|
||
{ capabilities: {} }
|
||
)
|
||
|
||
transport = new StreamableHTTPClientTransport(
|
||
new URL(endpoint),
|
||
{
|
||
requestInit: {
|
||
headers: {
|
||
'x-app-id': appId,
|
||
'x-app-secret': appSecret,
|
||
},
|
||
},
|
||
}
|
||
)
|
||
|
||
await client.connect(transport)
|
||
const toolsResult = await client.listTools()
|
||
const tools = toolsResult.tools || []
|
||
|
||
return { success: true, tools }
|
||
} catch (e: any) {
|
||
return { success: false, error: e.message }
|
||
} finally {
|
||
if (client) {
|
||
try {
|
||
await client.close()
|
||
} catch {}
|
||
}
|
||
}
|
||
}
|
||
|
||
async callTool(integrationId: string, toolName: string, args: any): Promise<any> {
|
||
const instance = this.mcpClients.get(integrationId)
|
||
if (!instance) {
|
||
throw new Error('MCP 未连接，请先连接')
|
||
}
|
||
|
||
const result = await instance.client.callTool({
|
||
name: toolName,
|
||
arguments: args,
|
||
})
|
||
|
||
return result
|
||
}
|
||
|
||
/**
|
||
* 通用 MCP 调用封装 - 带重试 + 状态返回
|
||
*
|
||
* @returns MCPResult<T> - 包含 success/data/error/latencyMs/attempts
|
||
*
|
||
* 失败重试策略：
|
||
* - 第一次失败后等待 1s 重试
|
||
* - 第二次失败后等待 2s 重试
|
||
* - 第三次失败后放弃
|
||
* - 如果未连接,自动尝试连接
|
||
*/
|
||
async executeTool<T = any>(
|
||
integrationId: string,
|
||
toolName: string,
|
||
args: any,
|
||
options: { retries?: number; retryDelayMs?: number; autoConnect?: boolean } = {},
|
||
): Promise<MCPResult<T>> {
|
||
const { retries = 2, retryDelayMs = 1000, autoConnect = true } = options
|
||
const startTime = Date.now()
|
||
let attempts = 0
|
||
let lastError = ''
|
||
|
||
for (let i = 0; i <= retries; i++) {
|
||
attempts = i + 1
|
||
try {
|
||
let instance = this.mcpClients.get(integrationId)
|
||
if (!instance && autoConnect) {
|
||
this.logger.log(`MCP 未连接 ${integrationId},尝试自动连接...`)
|
||
const connectResult = await this.connect(integrationId)
|
||
if (!connectResult.success) {
|
||
throw new Error(`自动连接失败: ${connectResult.error}`)
|
||
}
|
||
instance = this.mcpClients.get(integrationId)
|
||
}
|
||
if (!instance) {
|
||
throw new Error('MCP 客户端未初始化')
|
||
}
|
||
|
||
const result = await instance.client.callTool({ name: toolName, arguments: args })
|
||
const parsed = this.parseToolResult<T>(result)
|
||
|
||
return {
|
||
success: true,
|
||
data: parsed,
|
||
attempts,
|
||
latencyMs: Date.now() - startTime,
|
||
mode: 'mcp',
|
||
}
|
||
} catch (e: any) {
|
||
lastError = e.message || String(e)
|
||
this.logger.warn(`MCP tool 调用失败 (attempt ${attempts}/${retries + 1}): ${toolName} - ${lastError}`)
|
||
|
||
if (i < retries) {
|
||
await new Promise(resolve => setTimeout(resolve, retryDelayMs * (i + 1)))
|
||
}
|
||
}
|
||
}
|
||
|
||
return {
|
||
success: false,
|
||
error: lastError,
|
||
attempts,
|
||
latencyMs: Date.now() - startTime,
|
||
mode: 'local',
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 解析 MCP tool 返回的 content 字段
|
||
* MCP 协议格式: { content: [{ type: 'text', text: '...' }], isError?: boolean }
|
||
*/
|
||
private parseToolResult<T>(result: any): T {
|
||
if (!result) throw new Error('MCP 工具返回结果为空')
|
||
if (result.isError) {
|
||
const errMsg = typeof result.content?.[0]?.text === 'string' ? result.content[0].text : 'Tool 调用返回错误'
|
||
throw new Error(errMsg)
|
||
}
|
||
if (result.content && Array.isArray(result.content) && result.content.length > 0) {
|
||
const item = result.content[0]
|
||
if (item.type === 'text') {
|
||
try {
|
||
return JSON.parse(item.text) as T
|
||
} catch {
|
||
return item.text as unknown as T
|
||
}
|
||
}
|
||
}
|
||
return result as T
|
||
}
|
||
|
||
getClient(integrationId: string): Client | null {
|
||
const instance = this.mcpClients.get(integrationId)
|
||
return instance?.client || null
|
||
}
|
||
|
||
async findActive() {
|
||
return this.prisma.integration.findMany({
|
||
where: { status: 'active' },
|
||
})
|
||
}
|
||
|
||
/**
|
||
* 找到 status=active 或 error 的 MCP 集成（error 状态也包含,因为 executeTool 会自动重连）
|
||
*/
|
||
async findUsableMCP(): Promise<any[]> {
|
||
return this.prisma.integration.findMany({
|
||
where: {
|
||
type: 'mcp',
|
||
status: { in: ['active', 'error'] },
|
||
},
|
||
})
|
||
}
|
||
|
||
/**
|
||
* 找到第一个指定条件的活跃集成（按类型/名称）
|
||
*/
|
||
async findActiveByPredicate(predicate: (i: any) => boolean): Promise<any | null> {
|
||
const active = await this.findActive()
|
||
return active.find(predicate) || null
|
||
}
|
||
|
||
/**
|
||
* 找到第一个活跃的 MCP 类型集成(优先匹配名称包含 B2B/询价/platform_network,否则取第一个)
|
||
*/
|
||
async findActiveMCP(): Promise<any | null> {
|
||
const usable = await this.findUsableMCP()
|
||
if (usable.length === 0) return null
|
||
const priority = usable.find(i =>
|
||
i.type === 'mcp' && (
|
||
i.name.includes('B2B') ||
|
||
i.name.includes('询价') ||
|
||
i.name.includes('product_network') ||
|
||
i.name.includes('direv') ||
|
||
i.name.includes('巨变') ||
|
||
i.name.includes('平台')
|
||
)
|
||
)
|
||
return priority || usable.find(i => i.type === 'mcp') || usable[0]
|
||
}
|
||
}
|
||
|
||
export interface MCPResult<T = any> {
|
||
success: boolean
|
||
data?: T | null
|
||
error?: string
|
||
attempts: number
|
||
latencyMs: number
|
||
mode: 'mcp' | 'local'
|
||
}
|