import "server-only"; import { LoadAPIKeyError, UIMessage, Tool, jsonSchema, tool as createTool, isToolUIPart, UIMessagePart, ToolUIPart, getToolName, UIMessageStreamWriter, } from "ai"; import { ChatMention, ChatMetadata, ManualToolConfirmTag, } from "app-types/chat"; import { errorToString, exclude, objectFlow } from "lib/utils"; import logger from "logger"; import { AllowedMCPServer, McpServerCustomizationsPrompt, VercelAIMcpTool, VercelAIMcpToolTag, } from "app-types/mcp"; import { MANUAL_REJECT_RESPONSE_PROMPT } from "lib/ai/prompts"; import { ObjectJsonSchema7 } from "app-types/util"; import { safe } from "ts-safe"; import { workflowRepository } from "lib/db/repository"; import { VercelAIWorkflowTool, VercelAIWorkflowToolStreaming, VercelAIWorkflowToolStreamingResultTag, VercelAIWorkflowToolTag, } from "app-types/workflow"; import { createWorkflowExecutor } from "lib/ai/workflow/executor/workflow-executor"; import { NodeKind } from "lib/ai/workflow/workflow.interface"; import { mcpClientsManager } from "lib/ai/mcp/mcp-manager"; import { APP_DEFAULT_TOOL_KIT } from "lib/ai/tools/tool-kit"; import { AppDefaultToolkit } from "lib/ai/tools"; export function filterMCPToolsByMentions( tools: Record, mentions: ChatMention[], ) { if (mentions.length === 0) { return tools; } const toolMentions = mentions.filter( (mention) => mention.type == "mcpTool" || mention.type == "mcpServer", ) as Extract[]; const metionsByServer = toolMentions.reduce( (acc, mention) => { if (mention.type == "mcpServer") { return { ...acc, [mention.serverId]: Object.values(tools).map( (tool) => tool._originToolName, ), }; } return { ...acc, [mention.serverId]: [...(acc[mention.serverId] ?? []), mention.name], }; }, {} as Record, ); return objectFlow(tools).filter((_tool) => { if (!metionsByServer[_tool._mcpServerId]) return false; return metionsByServer[_tool._mcpServerId].includes(_tool._originToolName); }); } export function filterMCPToolsByAllowedMCPServers( tools: Record, allowedMcpServers?: Record, ): Record { if (!allowedMcpServers || Object.keys(allowedMcpServers).length === 0) { return {}; } return objectFlow(tools).filter((_tool) => { if (!allowedMcpServers[_tool._mcpServerId]?.tools) return false; return allowedMcpServers[_tool._mcpServerId].tools.includes( _tool._originToolName, ); }); } export function excludeToolExecution( tool: Record, ): Record { return objectFlow(tool).map((value) => { return createTool({ inputSchema: value.inputSchema, description: value.description, }); }); } export function mergeSystemPrompt( ...prompts: (string | undefined | false)[] ): string { const filteredPrompts = prompts .map((prompt) => (prompt ? prompt.trim() : "")) .filter(Boolean); return filteredPrompts.join("\n\n"); } export function manualToolExecuteByLastMessage( part: ToolUIPart, tools: Record, abortSignal?: AbortSignal, ) { const { input } = part; const toolName = getToolName(part); const tool = tools[toolName]; return safe(() => { if (!tool) throw new Error(`tool not found: ${toolName}`); if (!ManualToolConfirmTag.isMaybe(part.output)) throw new Error("manual tool confirm not found"); return part.output; }) .map(({ confirm }) => { if (!confirm) return MANUAL_REJECT_RESPONSE_PROMPT; if (VercelAIWorkflowToolTag.isMaybe(tool)) { return tool.execute!(input, { toolCallId: part.toolCallId, abortSignal: abortSignal ?? new AbortController().signal, messages: [], }); } else if (VercelAIMcpToolTag.isMaybe(tool)) { return mcpClientsManager.toolCall( tool._mcpServerId, tool._originToolName, input, ); } return tool.execute!(input, { toolCallId: part.toolCallId, abortSignal: abortSignal ?? new AbortController().signal, messages: [], }); }) .ifFail((error) => ({ isError: true, statusMessage: `tool call fail: ${toolName}`, error: errorToString(error), })) .unwrap(); } export function handleError(error: any) { if (LoadAPIKeyError.isInstance(error)) { return error.message; } logger.error(error); logger.error(`Route Error: ${error.name}`); return errorToString(error.message); } export function extractInProgressToolPart(message: UIMessage): ToolUIPart[] { if (message.role != "assistant") return []; if ((message.metadata as ChatMetadata)?.toolChoice != "manual") return []; return message.parts.filter( (part) => isToolUIPart(part) && part.state == "output-available" && ManualToolConfirmTag.isMaybe(part.output), ) as ToolUIPart[]; } export function filterMcpServerCustomizations( tools: Record, mcpServerCustomization: Record, ): Record { const toolNamesByServerId = Object.values(tools).reduce( (acc, tool) => { if (!acc[tool._mcpServerId]) acc[tool._mcpServerId] = []; acc[tool._mcpServerId].push(tool._originToolName); return acc; }, {} as Record, ); return Object.entries(mcpServerCustomization).reduce( (acc, [serverId, mcpServerCustomization]) => { if (!(serverId in toolNamesByServerId)) return acc; if ( !mcpServerCustomization.prompt && !Object.keys(mcpServerCustomization.tools ?? {}).length ) return acc; const prompts: McpServerCustomizationsPrompt = { id: serverId, name: mcpServerCustomization.name, prompt: mcpServerCustomization.prompt, tools: mcpServerCustomization.tools ? objectFlow(mcpServerCustomization.tools).filter((_, key) => { return toolNamesByServerId[serverId].includes(key as string); }) : {}, }; acc[serverId] = prompts; return acc; }, {} as Record, ); } export const workflowToVercelAITool = ({ id, description, schema, dataStream, name, }: { id: string; name: string; description?: string; schema: ObjectJsonSchema7; dataStream: UIMessageStreamWriter; }): VercelAIWorkflowTool => { const toolName = name .replace(/[^a-zA-Z0-9\s]/g, "") .trim() .replace(/\s+/g, "-") .toUpperCase(); const tool = createTool({ description: `${name} ${description?.trim().slice(0, 50)}`, inputSchema: jsonSchema(schema), execute(query, { toolCallId, abortSignal }) { const history: VercelAIWorkflowToolStreaming[] = []; const toolResult = VercelAIWorkflowToolStreamingResultTag.create({ toolCallId, workflowName: name, startedAt: Date.now(), endedAt: Date.now(), history, result: undefined, status: "running", }); return safe(id) .map((id) => workflowRepository.selectStructureById(id, { ignoreNote: true, }), ) .map((workflow) => { if (!workflow) throw new Error("Not Found Workflow"); const executor = createWorkflowExecutor({ nodes: workflow.nodes, edges: workflow.edges, }); toolResult.workflowIcon = workflow.icon; abortSignal?.addEventListener("abort", () => executor.exit()); executor.subscribe((e) => { if ( e.eventType == "WORKFLOW_START" || e.eventType == "WORKFLOW_END" ) return; if (e.node.name == "SKIP") return; if (e.eventType == "NODE_START") { const node = workflow.nodes.find( (node) => node.id == e.node.name, )!; if (!node) return; history.push({ id: e.nodeExecutionId, name: node.name, status: "running", startedAt: e.startedAt, kind: node.kind as NodeKind, }); } else if (e.eventType == "NODE_END") { const result = history.find((r) => r.id == e.nodeExecutionId); if (result) { if (e.isOk) { result.status = "success"; result.result = { input: e.node.output.getInput(e.node.name), output: e.node.output.getOutput({ nodeId: e.node.name, path: [], }), }; } else { result.status = "fail"; result.error = { name: e.error?.name || "ERROR", message: errorToString(e.error), }; } result.endedAt = e.endedAt; } } dataStream.write({ type: "tool-output-available", toolCallId, output: toolResult, }); }); return executor.run( { query: query ?? ({} as any), }, { disableHistory: true, }, ); }) .map((result) => { toolResult.endedAt = Date.now(); toolResult.status = result.isOk ? "success" : "fail"; toolResult.error = result.error ? { name: result.error.name || "ERROR", message: errorToString(result.error) || "Unknown Error", } : undefined; const outputNodeResults = history .filter((h) => h.kind == NodeKind.Output) .map((v) => v.result?.output) .filter(Boolean); toolResult.history = history.map((h) => ({ ...h, result: undefined, // save tokens. })); toolResult.result = outputNodeResults.length == 1 ? outputNodeResults[0] : outputNodeResults; return toolResult; }) .ifFail((err) => { return { error: { name: err?.name || "ERROR", message: errorToString(err), history, }, }; }) .unwrap(); }, }) as VercelAIWorkflowTool; tool._workflowId = id; tool._originToolName = name; tool._toolName = toolName; return VercelAIWorkflowToolTag.create(tool); }; export const workflowToVercelAITools = ( workflows: { id: string; name: string; description?: string; schema: ObjectJsonSchema7; }[], dataStream: UIMessageStreamWriter, ) => { return workflows .map((v) => workflowToVercelAITool({ ...v, dataStream, }), ) .reduce( (prev, cur) => { prev[cur._toolName] = cur; return prev; }, {} as Record, ); }; export const loadMcpTools = async (opt?: { mentions?: ChatMention[]; allowedMcpServers?: Record; }) => { try { const tools = await mcpClientsManager.tools(); if (opt?.mentions?.length) { return filterMCPToolsByMentions(tools, opt.mentions); } return filterMCPToolsByAllowedMCPServers(tools, opt?.allowedMcpServers); } catch (e) { return {} as Record; } }; export const loadWorkFlowTools = async (opt: { mentions?: ChatMention[]; dataStream: UIMessageStreamWriter; }) => { try { const tools = opt?.mentions?.length ? await workflowRepository.selectToolByIds( opt?.mentions ?.filter((m) => m.type == "workflow") .map( (v) => (v as Extract).workflowId, ), ) : []; return workflowToVercelAITools(tools, opt.dataStream); } catch (e) { return {} as Record; } }; export const loadAppDefaultTools = (opt?: { mentions?: ChatMention[]; allowedAppDefaultToolkit?: string[]; }) => safe(APP_DEFAULT_TOOL_KIT) .map((tools) => { if (opt?.mentions?.length) { const defaultToolMentions = opt.mentions.filter( (m) => m.type == "defaultTool", ); return Array.from(Object.values(tools)).reduce((acc, t) => { const allowed = objectFlow(t).filter((_, k) => { return defaultToolMentions.some((m) => m.name == k); }); return { ...acc, ...allowed }; }, {}); } const allowedAppDefaultToolkit = opt?.allowedAppDefaultToolkit ?? Object.values(AppDefaultToolkit); return ( allowedAppDefaultToolkit.reduce( (acc, key) => { return { ...acc, ...tools[key] }; }, {} as Record, ) || {} ); }) .ifFail((e) => { console.error(e); throw e; }) .orElse({} as Record); export const convertToSavePart = >( part: T, ) => { return safe( exclude(part as any, ["providerMetadata", "callProviderMetadata"]) as T, ) .map((v) => { if (isToolUIPart(v) && v.state.startsWith("output")) { if (VercelAIWorkflowToolStreamingResultTag.isMaybe(v.output)) { return { ...v, output: { ...v.output, history: v.output.history.map((h: any) => { return { ...h, result: undefined, }; }), }, }; } } return v; }) .unwrap(); };