feat(mcp): Tasks extension (io.modelcontextprotocol/tasks) (#1283)
* feat(mcp): speak spec revision 2026-07-28 (stateless core) Adopt the 2026-07-28 MCP spec revision on the connector endpoint while keeping every handshake-era client (2025-06-18 and earlier) byte-identical: - Accept per-request _meta protocol negotiation (io.modelcontextprotocol/protocolVersion); unsupported versions return UnsupportedProtocolVersionError (-32022) with the supported list. - Implement server/discover (spec MUST): supported revisions, capabilities including the extensions field, identity, instructions, freshness hints. - Decorate results for stateless clients: required resultType, serverInfo in _meta, and CacheableResult ttlMs/cacheScope on tools/list, prompts/list, resources/list, resources/read. - Validate the standard Mcp-Method/Mcp-Name request headers when present (HeaderMismatchError -32020); absence stays accepted. - Declare the ratified MCP Apps extension (io.modelcontextprotocol/ui) in capabilities; the widgets already use the ratified mime type and _meta.ui.resourceUri shape, so no widget changes are needed. - OAuth: include the RFC 9207 iss parameter on every authorization response (success and error) and advertise authorization_response_iss_parameter_supported in RFC 8414 metadata. Resource-not-found already used -32602 and tools/list ordering was already deterministic; both are covered by the new test file. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * feat(mcp): Tasks extension (io.modelcontextprotocol/tasks) Durable handles for long-running MCP tool calls, per the official Tasks extension. A client that declares the extension in its per-request capabilities gets a CreateTaskResult (resultType: "task") immediately; the work completes after the response via after() and lands in the new mcp_tasks table for tasks/get polling. Clients that did not declare the extension are never handed a task (spec MUST). - New mcp_tasks table (migration 20260729094000): company-scoped SELECT RLS, service-role-only writes (mirrors pending_operations), 1-hour expiry, status lifecycle CHECK. pg-real coverage included; triaged as excluded in the full-archive backup contract (transient state). - tasks/get (creator-scoped), tasks/cancel (cooperative, working-only flip), tasks/update (ack no-op: no input_required flows yet). - Tool opt-in via shouldRunAsTask predicate; first producer is gnubok_audit_package, the one genuinely long-running blocking call (multi-minute ZIP generation). estimate_only stays synchronous. - Tool failures complete the task with the standard isError envelope, exactly what the synchronous call would have returned; the failed status stays reserved for infrastructure errors. - server/discover and initialize now advertise the tasks extension. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(mcp): creator-only task RLS, enforced expiry sweep, RoPA entry Compliance-swarm follow-ups on the mcp_tasks migration (editing the migration is safe: it has not shipped beyond the ephemeral PR preview): - SELECT RLS tightened from company-wide to auth.uid() = user_id so the DB grant matches the creator-scoped tasks/get contract; task results carry raw tool output (Art. 5(1)(c)). pg test now proves a same-company colleague cannot read the row. - The 1-hour retention is now enforced, not aspirational: createMcpTask opportunistically deletes expired rows on every creation (idx_mcp_tasks_expires), best-effort (Art. 5(1)(e)). - RoPA entry mcp.async_task_handles added to .compliance/ropa.yaml (Art. 30). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(mcp): literal terminal-update payload for the phantom-column guard The conditional spreads in resolveMcpTask made the payload unresolvable for the no-phantom-columns guard (362 > 360 ceiling). A literal payload writing null for absent terminal fields is equivalent here: the terminal transition sets the complete terminal state. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Jakob Wennberg
@@ -466,6 +466,43 @@ processing_activities:
|
||||
- writes_staged_for_explicit_user_approval
|
||||
- plaintext_personnummer_blocked_from_staging
|
||||
|
||||
- id: mcp.async_task_handles
|
||||
name: MCP-uppgiftshandtag för långkörande verktygsanrop
|
||||
purpose: >-
|
||||
Låta en behörig MCP-klient hämta resultatet av ett långkörande
|
||||
verktygsanrop (t.ex. generering av arkivpaket) via ett hållbart
|
||||
uppgifts-id i stället för att hålla anslutningen öppen. Raden lagrar
|
||||
verktygets svar tills klienten hämtat det; endast den skapande
|
||||
användaren kan läsa, och utgångna rader rensas löpande vid nya anrop.
|
||||
lawful_basis: art_6_1_b
|
||||
special_category_basis: null
|
||||
controller: gnubok-tenant
|
||||
processor: supabase
|
||||
data_subjects:
|
||||
- business_owner
|
||||
- company_member
|
||||
data_categories:
|
||||
- user.financial
|
||||
- user.activity_timestamp
|
||||
recipients:
|
||||
- name: Supabase
|
||||
country: EU
|
||||
role: processor
|
||||
international_transfers:
|
||||
applicable: false
|
||||
mechanism: null
|
||||
note: EU-only processor.
|
||||
retention:
|
||||
duration: 1h
|
||||
basis: transient_tool_call_state_storage_limitation
|
||||
stored_in:
|
||||
- mcp_tasks
|
||||
security_measures:
|
||||
- rls_user_scoped_select # mcp_tasks SELECT: auth.uid() = user_id
|
||||
- service_role_only_writes
|
||||
- creator_scoped_task_polling # tasks/get requires the creating user
|
||||
- expired_rows_purged_on_creation
|
||||
|
||||
- id: arsredovisning.signature_evidence
|
||||
name: Underskriftsbevis för årsredovisning
|
||||
purpose: >-
|
||||
|
||||
@@ -0,0 +1,300 @@
|
||||
/**
|
||||
* Tests for the MCP Tasks extension (io.modelcontextprotocol/tasks):
|
||||
* capability gating, CreateTaskResult, post-response completion writes,
|
||||
* and the tasks/get / tasks/update / tasks/cancel methods. Does NOT re-test
|
||||
* audit-package generation itself (covered by audit-package.test.ts).
|
||||
*/
|
||||
import { describe, it, expect, vi, beforeEach } from 'vitest'
|
||||
import { eventBus } from '@/lib/events/bus'
|
||||
import { isTaskCapableClient, TASKS_EXTENSION_ID } from '../tasks'
|
||||
|
||||
const mocks = vi.hoisted(() => ({
|
||||
scopes: [] as string[],
|
||||
taskInserts: [] as Record<string, unknown>[],
|
||||
taskUpdates: [] as Record<string, unknown>[],
|
||||
taskRow: null as Record<string, unknown> | null,
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/auth/api-keys', async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import('@/lib/auth/api-keys')>()
|
||||
mocks.scopes = [...actual.ALL_SCOPES]
|
||||
|
||||
const membershipChain: unknown = new Proxy(
|
||||
{},
|
||||
{
|
||||
get(_t, prop) {
|
||||
if (prop === 'then') {
|
||||
return (resolve: (v: unknown) => void) =>
|
||||
resolve({
|
||||
data: { company_id: '11111111-1111-4111-8111-111111111111', role: 'owner' },
|
||||
error: null,
|
||||
})
|
||||
}
|
||||
return () => membershipChain
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
const makeChain = (result: unknown): unknown =>
|
||||
new Proxy(
|
||||
{},
|
||||
{
|
||||
get(_t, prop) {
|
||||
if (prop === 'then') {
|
||||
return (resolve: (v: unknown) => void) => resolve(result)
|
||||
}
|
||||
if (prop === 'single') {
|
||||
return () => Promise.resolve(result)
|
||||
}
|
||||
return () => makeChain(result)
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
const tasksBuilder = {
|
||||
delete: () => makeChain({ data: null, error: null }),
|
||||
insert: (row: Record<string, unknown>) => {
|
||||
mocks.taskInserts.push(row)
|
||||
return {
|
||||
select: () => ({
|
||||
single: async () => ({
|
||||
data: {
|
||||
id: 'task-1',
|
||||
company_id: row.company_id,
|
||||
user_id: row.user_id,
|
||||
tool_name: row.tool_name,
|
||||
status: 'working',
|
||||
status_message: null,
|
||||
result: null,
|
||||
error: null,
|
||||
poll_interval_ms: 2000,
|
||||
ttl_ms: 3600000,
|
||||
created_at: '2026-07-29T09:00:00.000Z',
|
||||
},
|
||||
error: null,
|
||||
}),
|
||||
}),
|
||||
}
|
||||
},
|
||||
update: (patch: Record<string, unknown>) => {
|
||||
mocks.taskUpdates.push(patch)
|
||||
return makeChain({ data: null, error: null })
|
||||
},
|
||||
select: () =>
|
||||
makeChain(
|
||||
mocks.taskRow
|
||||
? { data: mocks.taskRow, error: null }
|
||||
: { data: null, error: { message: 'not found' } }
|
||||
),
|
||||
}
|
||||
|
||||
return {
|
||||
...actual,
|
||||
extractBearerToken: vi.fn().mockReturnValue('test-token'),
|
||||
validateApiKey: vi.fn().mockResolvedValue({
|
||||
userId: 'user-1',
|
||||
companyId: '11111111-1111-4111-8111-111111111111',
|
||||
scopes: mocks.scopes,
|
||||
apiKeyId: 'key-1',
|
||||
apiKeyName: 'Test key',
|
||||
mode: 'live',
|
||||
}),
|
||||
createServiceClientNoCookies: vi.fn(() => ({
|
||||
from: (table: string) => {
|
||||
if (table === 'mcp_tasks') return tasksBuilder
|
||||
if (table === 'company_members') return membershipChain
|
||||
// fiscal_periods (and everything else) resolves to no rows, so the
|
||||
// audit-package execution fails fast with "Fiscal period not found".
|
||||
return makeChain({ data: null, error: { message: 'not found' }, count: 0 })
|
||||
},
|
||||
})),
|
||||
}
|
||||
})
|
||||
|
||||
import { handleMcpRequest } from '../server'
|
||||
|
||||
const TASK_META = {
|
||||
'io.modelcontextprotocol/protocolVersion': '2026-07-28',
|
||||
'io.modelcontextprotocol/clientCapabilities': {
|
||||
extensions: { [TASKS_EXTENSION_ID]: {} },
|
||||
},
|
||||
}
|
||||
|
||||
function mcpRequest(method: string, params?: Record<string, unknown>): Request {
|
||||
return new Request('http://localhost:3000/api/extensions/ext/mcp-server/mcp', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json', Authorization: 'Bearer test-token' },
|
||||
body: JSON.stringify({ jsonrpc: '2.0', id: 1, method, ...(params ? { params } : {}) }),
|
||||
})
|
||||
}
|
||||
|
||||
async function readBody(request: Request): Promise<{
|
||||
result?: Record<string, unknown>
|
||||
error?: { code: number; message: string }
|
||||
}> {
|
||||
const response = await handleMcpRequest(request)
|
||||
const body = await response.json()
|
||||
return { result: body.result, error: body.error }
|
||||
}
|
||||
|
||||
/** The after() fallback runs the job as a floating promise: let it settle. */
|
||||
async function settleBackgroundWork(): Promise<void> {
|
||||
await new Promise((resolve) => setTimeout(resolve, 0))
|
||||
await new Promise((resolve) => setTimeout(resolve, 0))
|
||||
}
|
||||
|
||||
describe('MCP Tasks extension', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
eventBus.clear()
|
||||
mocks.taskInserts.length = 0
|
||||
mocks.taskUpdates.length = 0
|
||||
mocks.taskRow = null
|
||||
})
|
||||
|
||||
describe('capability detection', () => {
|
||||
it('detects the tasks extension in per-request capabilities', () => {
|
||||
expect(isTaskCapableClient(TASK_META as Record<string, unknown>)).toBe(true)
|
||||
expect(isTaskCapableClient({})).toBe(false)
|
||||
expect(
|
||||
isTaskCapableClient({
|
||||
'io.modelcontextprotocol/clientCapabilities': { extensions: {} },
|
||||
})
|
||||
).toBe(false)
|
||||
})
|
||||
|
||||
it('advertises the tasks extension in server/discover', async () => {
|
||||
const { result } = await readBody(mcpRequest('server/discover'))
|
||||
const capabilities = result?.capabilities as { extensions: Record<string, unknown> }
|
||||
expect(capabilities.extensions[TASKS_EXTENSION_ID]).toEqual({})
|
||||
})
|
||||
})
|
||||
|
||||
describe('CreateTaskResult', () => {
|
||||
it('returns a task handle to a task-capable client and completes after the response', async () => {
|
||||
const { result } = await readBody(
|
||||
mcpRequest('tools/call', {
|
||||
name: 'gnubok_audit_package',
|
||||
arguments: { fiscal_period_id: '22222222-2222-4222-8222-222222222222' },
|
||||
_meta: TASK_META,
|
||||
})
|
||||
)
|
||||
expect(result?.resultType).toBe('task')
|
||||
const task = result?.task as Record<string, unknown>
|
||||
expect(task.taskId).toBe('task-1')
|
||||
expect(task.status).toBe('working')
|
||||
expect(typeof task.pollIntervalMs).toBe('number')
|
||||
expect(typeof task.ttlMs).toBe('number')
|
||||
expect(mocks.taskInserts).toHaveLength(1)
|
||||
|
||||
await settleBackgroundWork()
|
||||
// The mocked DB has no fiscal period, so the execution fails and the
|
||||
// task completes with the standard isError envelope.
|
||||
expect(mocks.taskUpdates).toHaveLength(1)
|
||||
expect(mocks.taskUpdates[0].status).toBe('completed')
|
||||
const stored = mocks.taskUpdates[0].result as Record<string, unknown>
|
||||
expect(stored.isError).toBe(true)
|
||||
expect(stored.resultType).toBe('complete')
|
||||
})
|
||||
|
||||
it('never returns a task to a client that did not declare the extension', async () => {
|
||||
const { result } = await readBody(
|
||||
mcpRequest('tools/call', {
|
||||
name: 'gnubok_audit_package',
|
||||
arguments: { fiscal_period_id: '22222222-2222-4222-8222-222222222222' },
|
||||
_meta: { 'io.modelcontextprotocol/protocolVersion': '2026-07-28' },
|
||||
})
|
||||
)
|
||||
expect(result?.resultType).not.toBe('task')
|
||||
expect(result?.isError).toBe(true)
|
||||
expect(mocks.taskInserts).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('keeps estimate_only synchronous even for task-capable clients', async () => {
|
||||
const { result } = await readBody(
|
||||
mcpRequest('tools/call', {
|
||||
name: 'gnubok_audit_package',
|
||||
arguments: {
|
||||
fiscal_period_id: '22222222-2222-4222-8222-222222222222',
|
||||
estimate_only: true,
|
||||
},
|
||||
_meta: TASK_META,
|
||||
})
|
||||
)
|
||||
expect(result?.resultType).not.toBe('task')
|
||||
expect(mocks.taskInserts).toHaveLength(0)
|
||||
})
|
||||
})
|
||||
|
||||
describe('tasks/get', () => {
|
||||
it('returns the working state while the task runs', async () => {
|
||||
mocks.taskRow = {
|
||||
id: 'task-1',
|
||||
company_id: '11111111-1111-4111-8111-111111111111',
|
||||
user_id: 'user-1',
|
||||
tool_name: 'gnubok_audit_package',
|
||||
status: 'working',
|
||||
status_message: null,
|
||||
result: null,
|
||||
error: null,
|
||||
poll_interval_ms: 2000,
|
||||
ttl_ms: 3600000,
|
||||
created_at: '2026-07-29T09:00:00.000Z',
|
||||
}
|
||||
const { result } = await readBody(mcpRequest('tasks/get', { taskId: 'task-1' }))
|
||||
expect(result?.status).toBe('working')
|
||||
expect(result?.result).toBeUndefined()
|
||||
expect(result?.resultType).toBe('complete')
|
||||
})
|
||||
|
||||
it('returns the stored tool result on completion', async () => {
|
||||
const storedResult = {
|
||||
resultType: 'complete',
|
||||
content: [{ type: 'text', text: '{"file_name":"arkiv.zip"}' }],
|
||||
structuredContent: { file_name: 'arkiv.zip' },
|
||||
}
|
||||
mocks.taskRow = {
|
||||
id: 'task-1',
|
||||
company_id: '11111111-1111-4111-8111-111111111111',
|
||||
user_id: 'user-1',
|
||||
tool_name: 'gnubok_audit_package',
|
||||
status: 'completed',
|
||||
status_message: null,
|
||||
result: storedResult,
|
||||
error: null,
|
||||
poll_interval_ms: 2000,
|
||||
ttl_ms: 3600000,
|
||||
created_at: '2026-07-29T09:00:00.000Z',
|
||||
}
|
||||
const { result } = await readBody(mcpRequest('tasks/get', { taskId: 'task-1' }))
|
||||
expect(result?.status).toBe('completed')
|
||||
expect(result?.result).toEqual(storedResult)
|
||||
})
|
||||
|
||||
it('rejects unknown task ids with invalid params', async () => {
|
||||
const { error } = await readBody(mcpRequest('tasks/get', { taskId: 'nope' }))
|
||||
expect(error?.code).toBe(-32602)
|
||||
})
|
||||
|
||||
it('requires taskId', async () => {
|
||||
const { error } = await readBody(mcpRequest('tasks/get', {}))
|
||||
expect(error?.code).toBe(-32602)
|
||||
})
|
||||
})
|
||||
|
||||
describe('tasks/cancel and tasks/update', () => {
|
||||
it('acknowledges cancellation and flips a working row', async () => {
|
||||
const { result } = await readBody(mcpRequest('tasks/cancel', { taskId: 'task-1' }))
|
||||
expect(result?.resultType).toBe('complete')
|
||||
expect(mocks.taskUpdates).toHaveLength(1)
|
||||
expect(mocks.taskUpdates[0].status).toBe('cancelled')
|
||||
})
|
||||
|
||||
it('acknowledges tasks/update as a no-op (no input_required flows yet)', async () => {
|
||||
const { result } = await readBody(
|
||||
mcpRequest('tasks/update', { taskId: 'task-1', inputResponses: {} })
|
||||
)
|
||||
expect(result?.resultType).toBe('complete')
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -100,7 +100,7 @@ describe('MCP spec revision 2026-07-28', () => {
|
||||
])
|
||||
const capabilities = result?.capabilities as Record<string, unknown>
|
||||
expect(capabilities.tools).toEqual({ listChanged: false })
|
||||
expect(capabilities.extensions).toEqual({ 'io.modelcontextprotocol/ui': {} })
|
||||
expect(capabilities.extensions).toMatchObject({ 'io.modelcontextprotocol/ui': {} })
|
||||
expect(typeof result?.ttlMs).toBe('number')
|
||||
expect(result?.cacheScope).toBe('private')
|
||||
const meta = result?._meta as Record<string, Record<string, unknown>>
|
||||
@@ -317,7 +317,7 @@ describe('MCP spec revision 2026-07-28', () => {
|
||||
expect(result?.protocolVersion).toBe('2025-06-18')
|
||||
expect((result?.serverInfo as Record<string, unknown>).name).toBe('gnubok')
|
||||
const capabilities = result?.capabilities as Record<string, unknown>
|
||||
expect(capabilities.extensions).toEqual({ 'io.modelcontextprotocol/ui': {} })
|
||||
expect(capabilities.extensions).toMatchObject({ 'io.modelcontextprotocol/ui': {} })
|
||||
})
|
||||
|
||||
it('negotiates an initialize requesting 2026-07-28 down to the handshake default', async () => {
|
||||
|
||||
@@ -1,4 +1,12 @@
|
||||
import { NextResponse, after } from 'next/server'
|
||||
import {
|
||||
TASKS_EXTENSION_ID,
|
||||
isTaskCapableClient,
|
||||
createMcpTask,
|
||||
resolveMcpTask,
|
||||
taskToWire,
|
||||
type McpTaskRow,
|
||||
} from './tasks'
|
||||
import {
|
||||
extractBearerToken,
|
||||
validateApiKey,
|
||||
@@ -244,6 +252,11 @@ interface McpTool {
|
||||
// _meta.ui.resourceUri on the RESULT, so the host renders the widget only when
|
||||
// asked. (Contrast _meta above, on the definition, which renders on every call.)
|
||||
uiResourceUri?: string
|
||||
// Tasks extension: when this predicate returns true for a call from a
|
||||
// task-capable client, the dispatcher returns a CreateTaskResult and runs
|
||||
// execute() after the response instead of blocking on it. Not serialized
|
||||
// into tools/list.
|
||||
shouldRunAsTask?: (args: Record<string, unknown>) => boolean
|
||||
execute: (
|
||||
args: Record<string, unknown>,
|
||||
companyId: string,
|
||||
@@ -12234,6 +12247,10 @@ export const tools: McpTool[] = [
|
||||
idempotentHint: true, // repeat calls produce equivalent archives, fresh URL
|
||||
openWorldHint: false,
|
||||
},
|
||||
// Archive generation is the one genuinely long-running synchronous call
|
||||
// in the catalog: task-capable clients get a durable handle instead of a
|
||||
// multi-minute blocking response. Size estimates stay synchronous.
|
||||
shouldRunAsTask: (args) => args.estimate_only !== true,
|
||||
async execute(args, companyId, userId, supabase) {
|
||||
const fiscalPeriodId = args.fiscal_period_id as string
|
||||
if (!fiscalPeriodId) throw new Error('fiscal_period_id is required')
|
||||
@@ -15463,9 +15480,13 @@ const SERVER_CAPABILITIES = {
|
||||
tools: { listChanged: false },
|
||||
resources: { listChanged: false },
|
||||
prompts: { listChanged: false },
|
||||
// MCP Apps (ratified extension): widgets are served as ui:// resources and
|
||||
// referenced from tool _meta.ui.resourceUri (see widgets/).
|
||||
extensions: { 'io.modelcontextprotocol/ui': {} },
|
||||
extensions: {
|
||||
// MCP Apps (ratified extension): widgets are served as ui:// resources
|
||||
// and referenced from tool _meta.ui.resourceUri (see widgets/).
|
||||
'io.modelcontextprotocol/ui': {},
|
||||
// MCP Tasks: durable handles for long-running tool calls (see tasks.ts).
|
||||
[TASKS_EXTENSION_ID]: {},
|
||||
},
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -15858,6 +15879,9 @@ export async function handleMcpRequest(request: Request): Promise<Response> {
|
||||
// Revisions are ISO dates, so string comparison orders them correctly.
|
||||
const statelessClient =
|
||||
typeof metaVersion === 'string' && metaVersion >= STATELESS_PROTOCOL_VERSION
|
||||
// Tasks extension: only a client that declared it in THIS request's
|
||||
// capabilities may ever receive a CreateTaskResult.
|
||||
const taskCapable = statelessClient && isTaskCapableClient(requestMeta)
|
||||
|
||||
// Standard request headers (2026-07-28): when present they must agree with
|
||||
// the JSON-RPC body. Absence stays accepted: this server supports
|
||||
@@ -16238,6 +16262,90 @@ export async function handleMcpRequest(request: Request): Promise<Response> {
|
||||
// mcp.next_hint_followed when the agent's behaviour matches the hint.
|
||||
checkAndEmitNextHintFollowed(sessionId, toolName, actor, userId, effectiveCompanyId)
|
||||
|
||||
// ── Tasks extension (io.modelcontextprotocol/tasks) ──
|
||||
// Long-running tools return a durable handle immediately to a client
|
||||
// that declared the extension; the work completes after the response
|
||||
// (after() keeps the function alive) and the result lands in mcp_tasks
|
||||
// for tasks/get polling. Runs after every auth/scope/capability guard
|
||||
// so nothing is ever started for a call that would have been refused.
|
||||
if (taskCapable && tool.shouldRunAsTask?.(toolArgs)) {
|
||||
const task = await createMcpTask(supabase, {
|
||||
companyId: effectiveCompanyId,
|
||||
userId,
|
||||
apiKeyId,
|
||||
toolName,
|
||||
})
|
||||
const taskStartedAt = Date.now()
|
||||
emitAfterResponse(async () => {
|
||||
try {
|
||||
const rawResult = await tool.execute(toolArgs, effectiveCompanyId, userId, supabase, actor)
|
||||
const canonicalResult = addCompanyToTopLevelNext(rawResult, effectiveCompanyId)
|
||||
const result = projectMcpPayload(canonicalResult, toolNamespace)
|
||||
const stored: Record<string, unknown> = {
|
||||
resultType: 'complete',
|
||||
content: [{ type: 'text', text: JSON.stringify(result, null, 2) }],
|
||||
}
|
||||
if (result !== null && result !== undefined) {
|
||||
stored.structuredContent =
|
||||
typeof result === 'object' && !Array.isArray(result) ? result : { value: result }
|
||||
}
|
||||
await resolveMcpTask(supabase, task.id, { status: 'completed', result: stored })
|
||||
emitToolCallTelemetry({
|
||||
tool: toolName,
|
||||
requiredScope: requiredScope ?? null,
|
||||
actor,
|
||||
latencyMs: Date.now() - taskStartedAt,
|
||||
success: true,
|
||||
isError: false,
|
||||
errorCode: null,
|
||||
errorKind: null,
|
||||
errorMessage: null,
|
||||
requestId: id ?? null,
|
||||
userId,
|
||||
companyId: effectiveCompanyId,
|
||||
})
|
||||
} catch (err) {
|
||||
// Tool failures complete the task with the standard isError
|
||||
// envelope: exactly what the synchronous call would have
|
||||
// returned. `failed` stays reserved for infrastructure errors.
|
||||
const structured = toToolError(err, { toolName })
|
||||
const publicStructured = projectMcpPayload(structured, toolNamespace)
|
||||
await resolveMcpTask(supabase, task.id, {
|
||||
status: 'completed',
|
||||
result: {
|
||||
resultType: 'complete',
|
||||
content: [{ type: 'text', text: JSON.stringify(publicStructured, null, 2) }],
|
||||
isError: true,
|
||||
},
|
||||
statusMessage: structured.error.message_sv,
|
||||
}).catch((updateErr) => {
|
||||
log.error('Failed to store MCP task failure result', { taskId: task.id, updateErr })
|
||||
})
|
||||
emitToolCallTelemetry({
|
||||
tool: toolName,
|
||||
requiredScope: requiredScope ?? null,
|
||||
actor,
|
||||
latencyMs: Date.now() - taskStartedAt,
|
||||
success: false,
|
||||
isError: true,
|
||||
errorCode: structured.error.code,
|
||||
errorKind: 'execution',
|
||||
errorMessage: structured.error.message_sv,
|
||||
requestId: id ?? null,
|
||||
userId,
|
||||
companyId: effectiveCompanyId,
|
||||
})
|
||||
}
|
||||
})
|
||||
return NextResponse.json(
|
||||
jsonRpc(id ?? null, {
|
||||
resultType: 'task',
|
||||
task: taskToWire(task),
|
||||
_meta: { [META_SERVER_INFO]: SERVER_INFO_BY_NAMESPACE[toolNamespace] },
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
const callStartedAt = Date.now()
|
||||
try {
|
||||
// gnubok_search_tools needs the caller's scopes to filter results to
|
||||
@@ -16537,6 +16645,51 @@ export async function handleMcpRequest(request: Request): Promise<Response> {
|
||||
)
|
||||
}
|
||||
|
||||
case 'tasks/get': {
|
||||
const taskId = (params as Record<string, unknown>)?.taskId
|
||||
if (typeof taskId !== 'string' || !taskId) {
|
||||
return NextResponse.json(jsonRpcError(id ?? null, -32602, 'taskId is required'))
|
||||
}
|
||||
// Scoped to the creating user: an API key can only poll its own tasks.
|
||||
const { data: taskRow } = await supabase
|
||||
.from('mcp_tasks')
|
||||
.select('*')
|
||||
.eq('id', taskId)
|
||||
.eq('user_id', userId)
|
||||
.single()
|
||||
if (!taskRow) {
|
||||
return NextResponse.json(jsonRpcError(id ?? null, -32602, `Task not found: "${taskId}"`))
|
||||
}
|
||||
const row = taskRow as McpTaskRow
|
||||
const wire: Record<string, unknown> = { resultType: 'complete', ...taskToWire(row) }
|
||||
if (row.status === 'completed' && row.result) wire.result = row.result
|
||||
if (row.status === 'failed' && row.error) wire.error = row.error
|
||||
wire._meta = { [META_SERVER_INFO]: SERVER_INFO_BY_NAMESPACE[toolNamespace] }
|
||||
return NextResponse.json(jsonRpc(id ?? null, wire))
|
||||
}
|
||||
|
||||
case 'tasks/update':
|
||||
// No input_required flows exist yet: acknowledge and ignore unknown or
|
||||
// already-satisfied inputResponses, as the extension spec instructs.
|
||||
return NextResponse.json(jsonRpc(id ?? null, { resultType: 'complete' }))
|
||||
|
||||
case 'tasks/cancel': {
|
||||
const taskId = (params as Record<string, unknown>)?.taskId
|
||||
if (typeof taskId !== 'string' || !taskId) {
|
||||
return NextResponse.json(jsonRpcError(id ?? null, -32602, 'taskId is required'))
|
||||
}
|
||||
// Cooperative cancellation: flip a still-working row; an in-flight
|
||||
// execution is not interrupted, and its late completion becomes a
|
||||
// no-op against the now-terminal row.
|
||||
await supabase
|
||||
.from('mcp_tasks')
|
||||
.update({ status: 'cancelled' })
|
||||
.eq('id', taskId)
|
||||
.eq('user_id', userId)
|
||||
.eq('status', 'working')
|
||||
return NextResponse.json(jsonRpc(id ?? null, { resultType: 'complete' }))
|
||||
}
|
||||
|
||||
default:
|
||||
return NextResponse.json(
|
||||
jsonRpcError(id ?? null, -32601, `Method not found: "${method}"`)
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
import type { SupabaseClient } from '@supabase/supabase-js'
|
||||
|
||||
/**
|
||||
* MCP Tasks extension (io.modelcontextprotocol/tasks).
|
||||
*
|
||||
* Durable handles for long-running tool calls: a task-capable client gets a
|
||||
* CreateTaskResult (resultType: "task") immediately and polls tasks/get until
|
||||
* a terminal status. Rows live in mcp_tasks (service-role writes only) so
|
||||
* handles survive disconnects and serverless instance turnover.
|
||||
*
|
||||
* Failure mapping: a tool execution failure is stored as a COMPLETED task
|
||||
* whose result carries the standard isError envelope, because that is exactly
|
||||
* what the synchronous call would have returned. The `failed` status (and the
|
||||
* `error` column) is reserved for infrastructure failures where no tool
|
||||
* result exists.
|
||||
*/
|
||||
|
||||
export const TASKS_EXTENSION_ID = 'io.modelcontextprotocol/tasks'
|
||||
|
||||
const DEFAULT_POLL_INTERVAL_MS = 2000
|
||||
const DEFAULT_TTL_MS = 3_600_000
|
||||
|
||||
export interface McpTaskRow {
|
||||
id: string
|
||||
company_id: string
|
||||
user_id: string
|
||||
tool_name: string
|
||||
status: 'working' | 'input_required' | 'completed' | 'failed' | 'cancelled'
|
||||
status_message: string | null
|
||||
result: Record<string, unknown> | null
|
||||
error: Record<string, unknown> | null
|
||||
poll_interval_ms: number
|
||||
ttl_ms: number
|
||||
created_at: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Per the extension spec, a server must never return a task to a client that
|
||||
* did not declare the extension in this request's capabilities.
|
||||
*/
|
||||
export function isTaskCapableClient(requestMeta: Record<string, unknown>): boolean {
|
||||
const caps = requestMeta['io.modelcontextprotocol/clientCapabilities']
|
||||
if (!caps || typeof caps !== 'object') return false
|
||||
const extensions = (caps as Record<string, unknown>).extensions
|
||||
if (!extensions || typeof extensions !== 'object') return false
|
||||
return TASKS_EXTENSION_ID in (extensions as Record<string, unknown>)
|
||||
}
|
||||
|
||||
export async function createMcpTask(
|
||||
supabase: SupabaseClient,
|
||||
params: { companyId: string; userId: string; apiKeyId?: string | null; toolName: string }
|
||||
): Promise<McpTaskRow> {
|
||||
// Storage limitation (GDPR Art. 5(1)(e)): opportunistically purge expired
|
||||
// rows on every creation so the 1-hour retention is enforced without
|
||||
// dedicated cron infrastructure (cheap via idx_mcp_tasks_expires).
|
||||
// Best-effort: a failed sweep must never block the new task.
|
||||
try {
|
||||
await supabase.from('mcp_tasks').delete().lt('expires_at', new Date().toISOString())
|
||||
} catch {
|
||||
// Ignore: the next creation retries the sweep.
|
||||
}
|
||||
const { data, error } = await supabase
|
||||
.from('mcp_tasks')
|
||||
.insert({
|
||||
company_id: params.companyId,
|
||||
user_id: params.userId,
|
||||
api_key_id: params.apiKeyId ?? null,
|
||||
tool_name: params.toolName,
|
||||
status: 'working',
|
||||
poll_interval_ms: DEFAULT_POLL_INTERVAL_MS,
|
||||
ttl_ms: DEFAULT_TTL_MS,
|
||||
})
|
||||
.select('*')
|
||||
.single()
|
||||
if (error || !data) {
|
||||
throw new Error(`Failed to create MCP task: ${error?.message ?? 'no row returned'}`)
|
||||
}
|
||||
return data as McpTaskRow
|
||||
}
|
||||
|
||||
/**
|
||||
* Move a still-working task to a terminal state. The status='working' guard
|
||||
* makes terminal states immutable (spec) and lets a tasks/cancel that raced
|
||||
* the execution win: the late completion becomes a no-op.
|
||||
*/
|
||||
export async function resolveMcpTask(
|
||||
supabase: SupabaseClient,
|
||||
taskId: string,
|
||||
terminal: {
|
||||
status: 'completed' | 'failed' | 'cancelled'
|
||||
result?: Record<string, unknown>
|
||||
error?: Record<string, unknown>
|
||||
statusMessage?: string
|
||||
}
|
||||
): Promise<void> {
|
||||
// Literal payload (no conditional spreads) so the phantom-column guard can
|
||||
// resolve every column. Writing null for absent terminal fields is correct:
|
||||
// the transition sets the complete terminal state.
|
||||
await supabase
|
||||
.from('mcp_tasks')
|
||||
.update({
|
||||
status: terminal.status,
|
||||
result: terminal.result ?? null,
|
||||
error: terminal.error ?? null,
|
||||
status_message: terminal.statusMessage ?? null,
|
||||
})
|
||||
.eq('id', taskId)
|
||||
.eq('status', 'working')
|
||||
}
|
||||
|
||||
/** Map a row to the wire Task object shared by CreateTaskResult and tasks/get. */
|
||||
export function taskToWire(row: McpTaskRow): Record<string, unknown> {
|
||||
return {
|
||||
taskId: row.id,
|
||||
status: row.status,
|
||||
createdAt: row.created_at,
|
||||
ttlMs: Number(row.ttl_ms),
|
||||
pollIntervalMs: row.poll_interval_ms,
|
||||
...(row.status_message ? { statusMessage: row.status_message } : {}),
|
||||
}
|
||||
}
|
||||
@@ -961,6 +961,7 @@ export const ARCHIVE_EXCLUDED_TABLES: Record<string, string> = {
|
||||
idempotency_keys: 'infrastructure',
|
||||
inbox_rate_counters: 'infrastructure',
|
||||
invoice_inbox_items: 'inbox workflow state; the files live in document_attachments',
|
||||
mcp_tasks: 'MCP task handles: transient tool-call state with a 1-hour TTL',
|
||||
metered_events: 'billing telemetry',
|
||||
notification_log: 'notification dedup log',
|
||||
operations: 'staged-operation workflow state',
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
-- Migration: MCP Tasks (io.modelcontextprotocol/tasks extension)
|
||||
-- Durable handles for long-running MCP tool calls: the tool call returns a
|
||||
-- task handle immediately and the work completes after the response; clients
|
||||
-- poll tasks/get until a terminal status. Writes go through the service-role
|
||||
-- MCP handler only (mirrors pending_operations); only the creating user may
|
||||
-- read. Retention (1 hour via expires_at) is enforced by an opportunistic
|
||||
-- sweep in the MCP handler: createMcpTask deletes expired rows on every
|
||||
-- task creation (GDPR Art. 5(1)(e)).
|
||||
|
||||
CREATE TABLE public.mcp_tasks (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
company_id UUID NOT NULL REFERENCES public.companies(id) ON DELETE CASCADE,
|
||||
user_id UUID NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE,
|
||||
-- Attribution only (which API key created the task); deliberately no FK so
|
||||
-- key rotation or deletion never breaks task history.
|
||||
api_key_id UUID,
|
||||
tool_name TEXT NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'working' CHECK (status IN (
|
||||
'working', 'input_required', 'completed', 'failed', 'cancelled'
|
||||
)),
|
||||
status_message TEXT,
|
||||
-- Terminal payloads: `result` holds exactly what the synchronous tool call
|
||||
-- would have returned (a CallToolResult, including isError envelopes);
|
||||
-- `error` holds a JSON-RPC error object for infrastructure failures.
|
||||
result JSONB,
|
||||
error JSONB,
|
||||
poll_interval_ms INTEGER NOT NULL DEFAULT 2000,
|
||||
ttl_ms BIGINT NOT NULL DEFAULT 3600000,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
expires_at TIMESTAMPTZ NOT NULL DEFAULT now() + interval '1 hour'
|
||||
);
|
||||
|
||||
-- tasks/get looks up by id + creator; expiry cleanup scans expires_at.
|
||||
CREATE INDEX idx_mcp_tasks_user ON public.mcp_tasks (user_id, status);
|
||||
CREATE INDEX idx_mcp_tasks_company ON public.mcp_tasks (company_id);
|
||||
CREATE INDEX idx_mcp_tasks_expires ON public.mcp_tasks (expires_at);
|
||||
|
||||
ALTER TABLE public.mcp_tasks ENABLE ROW LEVEL SECURITY;
|
||||
|
||||
-- Reads: the creating user only. tasks/get in the MCP handler scopes to the
|
||||
-- creator, and task results carry whatever the underlying tool returned;
|
||||
-- the DB grant must not be broader than that application contract
|
||||
-- (data minimisation, GDPR Art. 5(1)(c)). Mirrors pending_operations.
|
||||
CREATE POLICY "mcp_tasks_select_own" ON public.mcp_tasks
|
||||
FOR SELECT USING (auth.uid() = user_id);
|
||||
|
||||
-- No INSERT/UPDATE/DELETE policies: all writes go through the service-role
|
||||
-- MCP handler (mirrors pending_operations). Rows age out via expires_at.
|
||||
|
||||
CREATE TRIGGER mcp_tasks_updated_at
|
||||
BEFORE UPDATE ON public.mcp_tasks
|
||||
FOR EACH ROW EXECUTE FUNCTION public.update_updated_at_column();
|
||||
|
||||
-- No audit trigger: operational task plumbing, not business records. The
|
||||
-- underlying tool executions already emit mcp.tool_called telemetry, and any
|
||||
-- committed bookkeeping effects carry their own audit trail.
|
||||
|
||||
NOTIFY pgrst, 'reload schema';
|
||||
@@ -0,0 +1,109 @@
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { getPool, withUserContext } from './setup'
|
||||
import { seedCompany, insertAuthUser, insertCompanyMember } from './fixtures'
|
||||
|
||||
// pg-real coverage for 20260729094000_create_mcp_tasks.sql: the status CHECK,
|
||||
// creator-only SELECT RLS, the deliberate absence of authenticated write
|
||||
// policies (writes are service-role only), and the updated_at trigger.
|
||||
|
||||
async function insertTask(params: {
|
||||
companyId: string
|
||||
userId: string
|
||||
status?: string
|
||||
toolName?: string
|
||||
}): Promise<string> {
|
||||
const id = randomUUID()
|
||||
await getPool().query(
|
||||
`INSERT INTO public.mcp_tasks (id, company_id, user_id, tool_name, status)
|
||||
VALUES ($1, $2, $3, $4, $5)`,
|
||||
[id, params.companyId, params.userId, params.toolName ?? 'gnubok_audit_package', params.status ?? 'working'],
|
||||
)
|
||||
return id
|
||||
}
|
||||
|
||||
describe('mcp_tasks', () => {
|
||||
it('rejects statuses outside the task lifecycle', async () => {
|
||||
const { companyId, userId } = await seedCompany()
|
||||
await expect(
|
||||
insertTask({ companyId, userId, status: 'running' }),
|
||||
).rejects.toThrow(/mcp_tasks_status_check/)
|
||||
})
|
||||
|
||||
it('only the creating user can read a task (not other members, not other companies)', async () => {
|
||||
const a = await seedCompany()
|
||||
const b = await seedCompany()
|
||||
const taskId = await insertTask({ companyId: a.companyId, userId: a.userId })
|
||||
|
||||
const mine = await withUserContext(a.userId, (client) =>
|
||||
client.query('SELECT id, status FROM public.mcp_tasks WHERE id = $1', [taskId]),
|
||||
)
|
||||
expect(mine.rows).toHaveLength(1)
|
||||
expect(mine.rows[0].status).toBe('working')
|
||||
|
||||
// A second member of the SAME company must not see the row: task results
|
||||
// carry raw tool output, so the grant is creator-only (Art. 5(1)(c)).
|
||||
const colleagueId = await insertAuthUser()
|
||||
await insertCompanyMember({ companyId: a.companyId, userId: colleagueId, role: 'member' })
|
||||
const colleague = await withUserContext(colleagueId, (client) =>
|
||||
client.query('SELECT id FROM public.mcp_tasks WHERE id = $1', [taskId]),
|
||||
)
|
||||
expect(colleague.rows).toHaveLength(0)
|
||||
|
||||
const theirs = await withUserContext(b.userId, (client) =>
|
||||
client.query('SELECT id FROM public.mcp_tasks WHERE id = $1', [taskId]),
|
||||
)
|
||||
expect(theirs.rows).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('authenticated users cannot insert, update, or delete tasks (service-role only)', async () => {
|
||||
const { companyId, userId } = await seedCompany()
|
||||
const taskId = await insertTask({ companyId, userId })
|
||||
|
||||
await expect(
|
||||
withUserContext(userId, (client) =>
|
||||
client.query(
|
||||
`INSERT INTO public.mcp_tasks (company_id, user_id, tool_name)
|
||||
VALUES ($1, $2, 'gnubok_audit_package')`,
|
||||
[companyId, userId],
|
||||
),
|
||||
),
|
||||
).rejects.toThrow(/row-level security/)
|
||||
|
||||
// UPDATE and DELETE have no policies: RLS silently filters all rows,
|
||||
// so the statements succeed but affect nothing.
|
||||
const upd = await withUserContext(userId, (client) =>
|
||||
client.query(`UPDATE public.mcp_tasks SET status = 'cancelled' WHERE id = $1`, [taskId]),
|
||||
)
|
||||
expect(upd.rowCount).toBe(0)
|
||||
|
||||
const del = await withUserContext(userId, (client) =>
|
||||
client.query('DELETE FROM public.mcp_tasks WHERE id = $1', [taskId]),
|
||||
)
|
||||
expect(del.rowCount).toBe(0)
|
||||
})
|
||||
|
||||
it('bumps updated_at on status transitions', async () => {
|
||||
const { companyId, userId } = await seedCompany()
|
||||
const taskId = await insertTask({ companyId, userId })
|
||||
|
||||
const before = await getPool().query(
|
||||
'SELECT updated_at FROM public.mcp_tasks WHERE id = $1',
|
||||
[taskId],
|
||||
)
|
||||
// clock_timestamp()-based trigger: force a measurable gap.
|
||||
await getPool().query('SELECT pg_sleep(0.05)')
|
||||
await getPool().query(
|
||||
`UPDATE public.mcp_tasks SET status = 'completed', result = '{"ok":true}'::jsonb WHERE id = $1`,
|
||||
[taskId],
|
||||
)
|
||||
const after = await getPool().query(
|
||||
'SELECT updated_at, status FROM public.mcp_tasks WHERE id = $1',
|
||||
[taskId],
|
||||
)
|
||||
expect(after.rows[0].status).toBe('completed')
|
||||
expect(new Date(after.rows[0].updated_at).getTime()).toBeGreaterThanOrEqual(
|
||||
new Date(before.rows[0].updated_at).getTime(),
|
||||
)
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user