/**
* Odoo 工具執行器
*/
import { z } from 'zod'
import type { ToolResponse } from '@mcp-internal/shared'
import type { RecordLevelAccess } from '@mcp-internal/shared'
import { getOdooClient } from './client.js'
/** 允許的 Odoo 模型白名單 */
const ALLOWED_MODELS = [
'res.partner', 'res.users', 'res.company', 'res.currency',
'product.product', 'product.template', 'product.category',
'purchase.order', 'purchase.order.line',
'sale.order', 'sale.order.line',
'stock.picking', 'stock.move', 'stock.quant', 'stock.location', 'stock.warehouse',
'account.move', 'account.move.line', 'account.account', 'account.journal',
'account.analytic.account', 'account.analytic.line',
'project.project', 'project.task',
] as const
const DOMAIN_OPERATORS = [
'=', '!=', '>', '>=', '<', '<=',
'like', 'ilike', 'not like', 'not ilike',
'in', 'not in', '=like', '=ilike',
'child_of', 'parent_of',
] as const
const fieldNameSchema = z.string()
.min(1).max(100)
.regex(/^[a-zA-Z_][a-zA-Z0-9_.]*$/, '欄位名稱格式不正確')
const domainConditionSchema = z.tuple([
fieldNameSchema,
z.enum(DOMAIN_OPERATORS),
z.union([
z.string(), z.number(), z.boolean(), z.null(),
z.array(z.union([z.string(), z.number()])),
]),
])
const domainSchema = z.array(
z.union([domainConditionSchema, z.enum(['&', '|', '!'])])
).max(50, 'Domain 條件過多，最多 50 個')
const searchArgsSchema = z.object({
model: z.enum(ALLOWED_MODELS as unknown as [string, ...string[]]),
domain: domainSchema.optional().default([]),
fields: z.array(fieldNameSchema).max(100).optional(),
limit: z.number().int().min(1).max(1000).optional().default(20),
offset: z.number().int().min(0).max(100000).optional().default(0),
order: z.string().max(200).regex(/^[a-zA-Z_][a-zA-Z0-9_]*(\s+(asc|desc))?(,\s*[a-zA-Z_][a-zA-Z0-9_]*(\s+(asc|desc))?)*$/i).optional(),
})
const readArgsSchema = z.object({
model: z.enum(ALLOWED_MODELS as unknown as [string, ...string[]]),
ids: z.array(z.number().int().positive()).min(1).max(1000),
fields: z.array(fieldNameSchema).max(100).optional(),
})
const countArgsSchema = z.object({
model: z.enum(ALLOWED_MODELS as unknown as [string, ...string[]]),
domain: domainSchema.optional().default([]),
})
const createArgsSchema = z.object({
model: z.enum(ALLOWED_MODELS as unknown as [string, ...string[]]),
values: z.record(z.string(), z.unknown()).refine(
(val) => Object.keys(val).length > 0, { message: '值不能為空' }
),
})
const updateArgsSchema = z.object({
model: z.enum(ALLOWED_MODELS as unknown as [string, ...string[]]),
ids: z.array(z.number().int().positive()).min(1).max(100),
values: z.record(z.string(), z.unknown()).refine(
(val) => Object.keys(val).length > 0, { message: '值不能為空' }
),
})
const deleteArgsSchema = z.object({
model: z.enum(ALLOWED_MODELS as unknown as [string, ...string[]]),
ids: z.array(z.number().int().positive()).min(1).max(100),
})
/** 工具定義（供 Thin Client 使用） */
export const odooToolDefinitions = [
{
name: 'odoo_search',
description: '搜尋 Odoo 資料',
inputSchema: {
type: 'object',
properties: {
model: { type: 'string', description: 'Odoo 模型名稱' },
domain: { type: 'array', description: '搜尋條件' },
fields: { type: 'array', items: { type: 'string' }, description: '返回欄位列表' },
limit: { type: 'number', description: '返回數量限制', default: 20 },
offset: { type: 'number', description: '跳過記錄數', default: 0 },
order: { type: 'string', description: '排序方式' },
},
required: ['model'],
},
},
{
name: 'odoo_read',
description: '讀取指定 ID 的 Odoo 記錄',
inputSchema: {
type: 'object',
properties: {
model: { type: 'string', description: 'Odoo 模型名稱' },
ids: { type: 'array', items: { type: 'number' }, description: '記錄 ID 列表' },
fields: { type: 'array', items: { type: 'string' }, description: '返回欄位列表' },
},
required: ['model', 'ids'],
},
},
{
name: 'odoo_count',
description: '計算符合條件的 Odoo 記錄數',
inputSchema: {
type: 'object',
properties: {
model: { type: 'string', description: 'Odoo 模型名稱' },
domain: { type: 'array', description: '搜尋條件' },
},
required: ['model'],
},
},
{
name: 'odoo_create',
description: '建立 Odoo 記錄（僅 admin/assistant 可用）',
inputSchema: {
type: 'object',
properties: {
model: { type: 'string', description: 'Odoo 模型名稱' },
values: { type: 'object', description: '要建立的欄位和值' },
},
required: ['model', 'values'],
},
},
{
name: 'odoo_update',
description: '更新 Odoo 記錄（僅 admin/assistant 可用）',
inputSchema: {
type: 'object',
properties: {
model: { type: 'string', description: 'Odoo 模型名稱' },
ids: { type: 'array', items: { type: 'number' }, description: '要更新的記錄 ID' },
values: { type: 'object', description: '要更新的欄位和值' },
},
required: ['model', 'ids', 'values'],
},
},
{
name: 'odoo_delete',
description: '刪除 Odoo 記錄（僅 admin/assistant 可用）',
inputSchema: {
type: 'object',
properties: {
model: { type: 'string', description: 'Odoo 模型名稱' },
ids: { type: 'array', items: { type: 'number' }, description: '要刪除的記錄 ID' },
},
required: ['model', 'ids'],
},
},
]
function validateArgs(toolName: string, args: unknown): { success: true; data: any } | { success: false; error: string } {
try {
switch (toolName) {
case 'odoo_search': return { success: true, data: searchArgsSchema.parse(args) }
case 'odoo_read': return { success: true, data: readArgsSchema.parse(args) }
case 'odoo_count': return { success: true, data: countArgsSchema.parse(args) }
case 'odoo_create': return { success: true, data: createArgsSchema.parse(args) }
case 'odoo_update': return { success: true, data: updateArgsSchema.parse(args) }
case 'odoo_delete': return { success: true, data: deleteArgsSchema.parse(args) }
default: return { success: false, error: `未知的 Odoo 工具: ${toolName}` }
}
} catch (error) {
if (error instanceof z.ZodError) {
const messages = error.errors.map(e => `${e.path.join('.')}: ${e.message}`)
return { success: false, error: `參數驗證失敗: ${messages.join('; ')}` }
}
return { success: false, error: `參數驗證失敗: ${String(error)}` }
}
}
export async function executeOdooTool(
toolName: string,
args: unknown,
recordAccess: RecordLevelAccess
): Promise<ToolResponse> {
const startTime = Date.now()
const validation = validateArgs(toolName, args)
if (!validation.success) {
return {
success: false,
error: { code: 'VALIDATION_ERROR', message: validation.error },
metadata: { executionTimeMs: Date.now() - startTime, tool: toolName, timestamp: new Date().toISOString() },
}
}
const validatedArgs = validation.data
try {
const client = await getOdooClient()
let result: any
switch (toolName) {
case 'odoo_search':
if (validatedArgs.model === 'account.analytic.account' && recordAccess !== 'all') {
let domain: any[] = [['plan_id', '=', 6]]
if (validatedArgs.domain?.length > 0) {
domain = ['&', ...domain, ...validatedArgs.domain]
}
result = await client.search({ ...validatedArgs, domain })
} else {
result = await client.search(validatedArgs)
}
break
case 'odoo_read':
result = await client.read(validatedArgs)
break
case 'odoo_count':
result = await client.count(validatedArgs.model, validatedArgs.domain)
break
case 'odoo_create':
result = await client.create(validatedArgs)
break
case 'odoo_update':
result = await client.update(validatedArgs)
break
case 'odoo_delete':
result = await client.delete(validatedArgs)
break
default:
throw new Error(`未知的 Odoo 工具: ${toolName}`)
}
return {
success: true,
result: { content: [{ type: 'text', text: JSON.stringify(result, null, 2) }] },
metadata: { executionTimeMs: Date.now() - startTime, tool: toolName, timestamp: new Date().toISOString() },
}
} catch (error) {
return {
success: false,
error: { code: 'EXTERNAL_SERVICE_ERROR', message: error instanceof Error ? error.message : String(error) },
metadata: { executionTimeMs: Date.now() - startTime, tool: toolName, timestamp: new Date().toISOString() },
}
}
}
