1664 lines
65 KiB
TypeScript
1664 lines
65 KiB
TypeScript
import { createServer, type Server } from 'node:http';
|
|
import { randomUUID } from 'node:crypto';
|
|
import { Client } from '../../src/client/index.js';
|
|
import { StreamableHTTPClientTransport } from '../../src/client/streamableHttp.js';
|
|
import { McpServer } from '../../src/server/mcp.js';
|
|
import { StreamableHTTPServerTransport } from '../../src/server/streamableHttp.js';
|
|
import {
|
|
CallToolResultSchema,
|
|
CreateTaskResultSchema,
|
|
ElicitRequestSchema,
|
|
ElicitResultSchema,
|
|
ErrorCode,
|
|
McpError,
|
|
RELATED_TASK_META_KEY,
|
|
TaskSchema
|
|
} from '../../src/types.js';
|
|
import { z } from 'zod';
|
|
import { InMemoryTaskStore, InMemoryTaskMessageQueue } from '../../src/experimental/tasks/stores/in-memory.js';
|
|
import type { TaskRequestOptions } from '../../src/shared/protocol.js';
|
|
import { listenOnRandomPort } from '../helpers/http.js';
|
|
import { waitForTaskStatus } from '../helpers/tasks.js';
|
|
|
|
describe('Task Lifecycle Integration Tests', () => {
|
|
let server: Server;
|
|
let mcpServer: McpServer;
|
|
let serverTransport: StreamableHTTPServerTransport;
|
|
let baseUrl: URL;
|
|
let taskStore: InMemoryTaskStore;
|
|
|
|
beforeEach(async () => {
|
|
// Create task store
|
|
taskStore = new InMemoryTaskStore();
|
|
|
|
// Create MCP server with task support
|
|
mcpServer = new McpServer(
|
|
{ name: 'test-server', version: '1.0.0' },
|
|
{
|
|
capabilities: {
|
|
tasks: {
|
|
requests: {
|
|
tools: {
|
|
call: {}
|
|
}
|
|
},
|
|
list: {},
|
|
cancel: {}
|
|
}
|
|
},
|
|
taskStore,
|
|
taskMessageQueue: new InMemoryTaskMessageQueue()
|
|
}
|
|
);
|
|
|
|
// Register a long-running tool using registerToolTask
|
|
mcpServer.experimental.tasks.registerToolTask(
|
|
'long-task',
|
|
{
|
|
title: 'Long Running Task',
|
|
description: 'A tool that takes time to complete',
|
|
inputSchema: {
|
|
duration: z.number().describe('Duration in milliseconds').default(1000),
|
|
shouldFail: z.boolean().describe('Whether the task should fail').default(false)
|
|
}
|
|
},
|
|
{
|
|
async createTask({ duration, shouldFail }, extra) {
|
|
const task = await extra.taskStore.createTask({
|
|
ttl: 60000,
|
|
pollInterval: 100
|
|
});
|
|
|
|
// Simulate async work
|
|
(async () => {
|
|
await new Promise(resolve => setTimeout(resolve, duration));
|
|
|
|
try {
|
|
if (shouldFail) {
|
|
await extra.taskStore.storeTaskResult(task.taskId, 'failed', {
|
|
content: [{ type: 'text', text: 'Task failed as requested' }],
|
|
isError: true
|
|
});
|
|
} else {
|
|
await extra.taskStore.storeTaskResult(task.taskId, 'completed', {
|
|
content: [{ type: 'text', text: `Completed after ${duration}ms` }]
|
|
});
|
|
}
|
|
} catch {
|
|
// Task may have been cleaned up if test ended
|
|
}
|
|
})();
|
|
|
|
return { task };
|
|
},
|
|
async getTask(_args, extra) {
|
|
const task = await extra.taskStore.getTask(extra.taskId);
|
|
if (!task) {
|
|
throw new Error(`Task ${extra.taskId} not found`);
|
|
}
|
|
return task;
|
|
},
|
|
async getTaskResult(_args, extra) {
|
|
const result = await extra.taskStore.getTaskResult(extra.taskId);
|
|
return result as { content: Array<{ type: 'text'; text: string }> };
|
|
}
|
|
}
|
|
);
|
|
|
|
// Register a tool that requires input via elicitation
|
|
mcpServer.experimental.tasks.registerToolTask(
|
|
'input-task',
|
|
{
|
|
title: 'Input Required Task',
|
|
description: 'A tool that requires user input',
|
|
inputSchema: {
|
|
userName: z.string().describe('User name').optional()
|
|
}
|
|
},
|
|
{
|
|
async createTask({ userName }, extra) {
|
|
const task = await extra.taskStore.createTask({
|
|
ttl: 60000,
|
|
pollInterval: 100
|
|
});
|
|
|
|
// Perform async work that requires elicitation
|
|
(async () => {
|
|
await new Promise(resolve => setTimeout(resolve, 100));
|
|
|
|
// If userName not provided, request it via elicitation
|
|
if (!userName) {
|
|
const elicitationResult = await extra.sendRequest(
|
|
{
|
|
method: 'elicitation/create',
|
|
params: {
|
|
mode: 'form',
|
|
message: 'What is your name?',
|
|
requestedSchema: {
|
|
type: 'object',
|
|
properties: {
|
|
userName: { type: 'string' }
|
|
},
|
|
required: ['userName']
|
|
}
|
|
}
|
|
},
|
|
ElicitResultSchema,
|
|
{ relatedTask: { taskId: task.taskId } } as unknown as TaskRequestOptions
|
|
);
|
|
|
|
// Complete with the elicited name
|
|
const name =
|
|
elicitationResult.action === 'accept' && elicitationResult.content
|
|
? elicitationResult.content.userName
|
|
: 'Unknown';
|
|
try {
|
|
await extra.taskStore.storeTaskResult(task.taskId, 'completed', {
|
|
content: [{ type: 'text', text: `Hello, ${name}!` }]
|
|
});
|
|
} catch {
|
|
// Task may have been cleaned up if test ended
|
|
}
|
|
} else {
|
|
// Complete immediately if userName was provided
|
|
try {
|
|
await extra.taskStore.storeTaskResult(task.taskId, 'completed', {
|
|
content: [{ type: 'text', text: `Hello, ${userName}!` }]
|
|
});
|
|
} catch {
|
|
// Task may have been cleaned up if test ended
|
|
}
|
|
}
|
|
})();
|
|
|
|
return { task };
|
|
},
|
|
async getTask(_args, extra) {
|
|
const task = await extra.taskStore.getTask(extra.taskId);
|
|
if (!task) {
|
|
throw new Error(`Task ${extra.taskId} not found`);
|
|
}
|
|
return task;
|
|
},
|
|
async getTaskResult(_args, extra) {
|
|
const result = await extra.taskStore.getTaskResult(extra.taskId);
|
|
return result as { content: Array<{ type: 'text'; text: string }> };
|
|
}
|
|
}
|
|
);
|
|
|
|
// Create transport
|
|
serverTransport = new StreamableHTTPServerTransport({
|
|
sessionIdGenerator: () => randomUUID()
|
|
});
|
|
|
|
await mcpServer.connect(serverTransport);
|
|
|
|
// Create HTTP server
|
|
server = createServer(async (req, res) => {
|
|
await serverTransport.handleRequest(req, res);
|
|
});
|
|
|
|
// Start server
|
|
baseUrl = await listenOnRandomPort(server);
|
|
});
|
|
|
|
afterEach(async () => {
|
|
taskStore.cleanup();
|
|
await mcpServer.close().catch(() => {});
|
|
await serverTransport.close().catch(() => {});
|
|
server.close();
|
|
});
|
|
|
|
describe('Task Creation and Completion', () => {
|
|
it('should create a task and return CreateTaskResult', async () => {
|
|
const client = new Client({
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Create a task
|
|
const createResult = await client.request(
|
|
{
|
|
method: 'tools/call',
|
|
params: {
|
|
name: 'long-task',
|
|
arguments: {
|
|
duration: 500,
|
|
shouldFail: false
|
|
},
|
|
task: {
|
|
ttl: 60000
|
|
}
|
|
}
|
|
},
|
|
CreateTaskResultSchema
|
|
);
|
|
|
|
// Verify CreateTaskResult structure
|
|
expect(createResult).toHaveProperty('task');
|
|
expect(createResult.task).toHaveProperty('taskId');
|
|
expect(createResult.task.status).toBe('working');
|
|
expect(createResult.task.ttl).toBe(60000);
|
|
expect(createResult.task.createdAt).toBeDefined();
|
|
expect(createResult.task.pollInterval).toBe(100);
|
|
|
|
// Verify task is stored in taskStore
|
|
const taskId = createResult.task.taskId;
|
|
const storedTask = await taskStore.getTask(taskId);
|
|
expect(storedTask).toBeDefined();
|
|
expect(storedTask?.taskId).toBe(taskId);
|
|
expect(storedTask?.status).toBe('working');
|
|
|
|
// Wait for completion
|
|
const completedTask = await waitForTaskStatus(id => taskStore.getTask(id), taskId, 'completed');
|
|
|
|
// Verify task completed
|
|
expect(completedTask.status).toBe('completed');
|
|
|
|
// Verify result is stored
|
|
const result = await taskStore.getTaskResult(taskId);
|
|
expect(result).toBeDefined();
|
|
expect(result.content).toEqual([{ type: 'text', text: 'Completed after 500ms' }]);
|
|
|
|
await transport.close();
|
|
});
|
|
|
|
it('should handle task failure correctly', async () => {
|
|
const client = new Client({
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Create a task that will fail
|
|
const createResult = await client.request(
|
|
{
|
|
method: 'tools/call',
|
|
params: {
|
|
name: 'long-task',
|
|
arguments: {
|
|
duration: 300,
|
|
shouldFail: true
|
|
},
|
|
task: {
|
|
ttl: 60000
|
|
}
|
|
}
|
|
},
|
|
CreateTaskResultSchema
|
|
);
|
|
|
|
const taskId = createResult.task.taskId;
|
|
|
|
// Wait for failure
|
|
const task = await waitForTaskStatus(id => taskStore.getTask(id), taskId, 'failed');
|
|
|
|
// Verify task failed
|
|
expect(task.status).toBe('failed');
|
|
|
|
// Verify error result is stored
|
|
const result = await taskStore.getTaskResult(taskId);
|
|
expect(result.content).toEqual([{ type: 'text', text: 'Task failed as requested' }]);
|
|
expect(result.isError).toBe(true);
|
|
|
|
await transport.close();
|
|
});
|
|
});
|
|
|
|
describe('Task Cancellation', () => {
|
|
it('should cancel a working task and return the cancelled task', async () => {
|
|
const client = new Client({
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Create a long-running task
|
|
const createResult = await client.request(
|
|
{
|
|
method: 'tools/call',
|
|
params: {
|
|
name: 'long-task',
|
|
arguments: {
|
|
duration: 5000
|
|
},
|
|
task: {
|
|
ttl: 60000
|
|
}
|
|
}
|
|
},
|
|
CreateTaskResultSchema
|
|
);
|
|
|
|
const taskId = createResult.task.taskId;
|
|
|
|
// Verify task is working
|
|
let task = await taskStore.getTask(taskId);
|
|
expect(task?.status).toBe('working');
|
|
|
|
// Cancel the task via client.experimental.tasks.cancelTask - per spec, returns Result & Task
|
|
const cancelResult = await client.experimental.tasks.cancelTask(taskId);
|
|
|
|
// Verify the cancel response includes the cancelled task (per MCP spec CancelTaskResult is Result & Task)
|
|
expect(cancelResult.taskId).toBe(taskId);
|
|
expect(cancelResult.status).toBe('cancelled');
|
|
expect(cancelResult.createdAt).toBeDefined();
|
|
expect(cancelResult.lastUpdatedAt).toBeDefined();
|
|
expect(cancelResult.ttl).toBeDefined();
|
|
|
|
// Verify task is cancelled in store as well
|
|
task = await taskStore.getTask(taskId);
|
|
expect(task?.status).toBe('cancelled');
|
|
|
|
await transport.close();
|
|
});
|
|
|
|
it('should reject cancellation of completed task with error code -32602', async () => {
|
|
const client = new Client({
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Create a quick task
|
|
const createResult = await client.request(
|
|
{
|
|
method: 'tools/call',
|
|
params: {
|
|
name: 'long-task',
|
|
arguments: {
|
|
duration: 100
|
|
},
|
|
task: {
|
|
ttl: 60000
|
|
}
|
|
}
|
|
},
|
|
CreateTaskResultSchema
|
|
);
|
|
|
|
const taskId = createResult.task.taskId;
|
|
|
|
// Wait for completion
|
|
const task = await waitForTaskStatus(id => taskStore.getTask(id), taskId, 'completed');
|
|
|
|
// Verify task is completed
|
|
expect(task.status).toBe('completed');
|
|
|
|
// Try to cancel via tasks/cancel request (should fail with -32602)
|
|
await expect(client.experimental.tasks.cancelTask(taskId)).rejects.toSatisfy((error: McpError) => {
|
|
expect(error).toBeInstanceOf(McpError);
|
|
expect(error.code).toBe(ErrorCode.InvalidParams);
|
|
expect(error.message).toContain('Cannot cancel task in terminal status');
|
|
return true;
|
|
});
|
|
|
|
await transport.close();
|
|
});
|
|
});
|
|
|
|
describe('Multiple Queued Messages', () => {
|
|
it('should deliver multiple queued messages in order', async () => {
|
|
// Register a tool that sends multiple server requests during execution
|
|
mcpServer.experimental.tasks.registerToolTask(
|
|
'multi-request-task',
|
|
{
|
|
title: 'Multi Request Task',
|
|
description: 'A tool that sends multiple server requests',
|
|
inputSchema: {
|
|
requestCount: z.number().describe('Number of requests to send').default(3)
|
|
}
|
|
},
|
|
{
|
|
async createTask({ requestCount }, extra) {
|
|
const task = await extra.taskStore.createTask({
|
|
ttl: 60000,
|
|
pollInterval: 100
|
|
});
|
|
|
|
// Perform async work that sends multiple requests
|
|
(async () => {
|
|
await new Promise(resolve => setTimeout(resolve, 100));
|
|
|
|
const responses: string[] = [];
|
|
|
|
// Send multiple elicitation requests
|
|
for (let i = 0; i < requestCount; i++) {
|
|
const elicitationResult = await extra.sendRequest(
|
|
{
|
|
method: 'elicitation/create',
|
|
params: {
|
|
mode: 'form',
|
|
message: `Request ${i + 1} of ${requestCount}`,
|
|
requestedSchema: {
|
|
type: 'object',
|
|
properties: {
|
|
response: { type: 'string' }
|
|
},
|
|
required: ['response']
|
|
}
|
|
}
|
|
},
|
|
ElicitResultSchema,
|
|
{ relatedTask: { taskId: task.taskId } } as unknown as TaskRequestOptions
|
|
);
|
|
|
|
if (elicitationResult.action === 'accept' && elicitationResult.content) {
|
|
responses.push(elicitationResult.content.response as string);
|
|
}
|
|
}
|
|
|
|
// Complete with all responses
|
|
try {
|
|
await extra.taskStore.storeTaskResult(task.taskId, 'completed', {
|
|
content: [{ type: 'text', text: `Received responses: ${responses.join(', ')}` }]
|
|
});
|
|
} catch {
|
|
// Task may have been cleaned up if test ended
|
|
}
|
|
})();
|
|
|
|
return { task };
|
|
},
|
|
async getTask(_args, extra) {
|
|
const task = await extra.taskStore.getTask(extra.taskId);
|
|
if (!task) {
|
|
throw new Error(`Task ${extra.taskId} not found`);
|
|
}
|
|
return task;
|
|
},
|
|
async getTaskResult(_args, extra) {
|
|
const result = await extra.taskStore.getTaskResult(extra.taskId);
|
|
return result as { content: Array<{ type: 'text'; text: string }> };
|
|
}
|
|
}
|
|
);
|
|
|
|
const client = new Client(
|
|
{
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
},
|
|
{
|
|
capabilities: {
|
|
elicitation: {}
|
|
}
|
|
}
|
|
);
|
|
|
|
const receivedMessages: Array<{ method: string; message: string }> = [];
|
|
|
|
// Set up elicitation handler on client to track message order
|
|
client.setRequestHandler(ElicitRequestSchema, async request => {
|
|
// Track the message
|
|
receivedMessages.push({
|
|
method: request.method,
|
|
message: request.params.message
|
|
});
|
|
|
|
// Extract the request number from the message
|
|
const match = request.params.message.match(/Request (\d+) of (\d+)/);
|
|
const requestNum = match ? match[1] : 'unknown';
|
|
|
|
// Respond with the request number
|
|
return {
|
|
action: 'accept' as const,
|
|
content: {
|
|
response: `Response ${requestNum}`
|
|
}
|
|
};
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Create a task that will send 3 requests
|
|
const createResult = await client.request(
|
|
{
|
|
method: 'tools/call',
|
|
params: {
|
|
name: 'multi-request-task',
|
|
arguments: {
|
|
requestCount: 3
|
|
},
|
|
task: {
|
|
ttl: 60000
|
|
}
|
|
}
|
|
},
|
|
CreateTaskResultSchema
|
|
);
|
|
|
|
const taskId = createResult.task.taskId;
|
|
|
|
// Wait for messages to be queued
|
|
await new Promise(resolve => setTimeout(resolve, 200));
|
|
|
|
// Call tasks/result to receive all queued messages
|
|
// This should deliver all 3 elicitation requests in order
|
|
const result = await client.request(
|
|
{
|
|
method: 'tasks/result',
|
|
params: { taskId }
|
|
},
|
|
CallToolResultSchema
|
|
);
|
|
|
|
// Verify all messages were delivered in order
|
|
expect(receivedMessages.length).toBe(3);
|
|
expect(receivedMessages[0].message).toBe('Request 1 of 3');
|
|
expect(receivedMessages[1].message).toBe('Request 2 of 3');
|
|
expect(receivedMessages[2].message).toBe('Request 3 of 3');
|
|
|
|
// Verify final result includes all responses
|
|
expect(result.content).toEqual([{ type: 'text', text: 'Received responses: Response 1, Response 2, Response 3' }]);
|
|
|
|
// Verify task is completed
|
|
const task = await client.request(
|
|
{
|
|
method: 'tasks/get',
|
|
params: { taskId }
|
|
},
|
|
TaskSchema
|
|
);
|
|
expect(task.status).toBe('completed');
|
|
|
|
await transport.close();
|
|
}, 10000);
|
|
});
|
|
|
|
describe('Input Required Flow', () => {
|
|
it('should handle elicitation during tool execution', async () => {
|
|
// Complete flow phases:
|
|
// 1. Client creates task
|
|
// 2. Server queues elicitation request and sets status to input_required
|
|
// 3. Client polls tasks/get, sees input_required status
|
|
// 4. Client calls tasks/result to dequeue elicitation request
|
|
// 5. Client responds to elicitation
|
|
// 6. Server receives response, completes task
|
|
// 7. Client receives final result
|
|
|
|
const elicitClient = new Client(
|
|
{
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
},
|
|
{
|
|
capabilities: {
|
|
elicitation: {}
|
|
}
|
|
}
|
|
);
|
|
|
|
// Track elicitation request receipt
|
|
let elicitationReceived = false;
|
|
let elicitationRequestMeta: Record<string, unknown> | undefined;
|
|
|
|
// Set up elicitation handler on client
|
|
elicitClient.setRequestHandler(ElicitRequestSchema, async request => {
|
|
elicitationReceived = true;
|
|
elicitationRequestMeta = request.params._meta;
|
|
|
|
return {
|
|
action: 'accept' as const,
|
|
content: {
|
|
userName: 'TestUser'
|
|
}
|
|
};
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await elicitClient.connect(transport);
|
|
|
|
// Phase 1: Create task
|
|
const createResult = await elicitClient.request(
|
|
{
|
|
method: 'tools/call',
|
|
params: {
|
|
name: 'input-task',
|
|
arguments: {},
|
|
task: {
|
|
ttl: 60000
|
|
}
|
|
}
|
|
},
|
|
CreateTaskResultSchema
|
|
);
|
|
|
|
const taskId = createResult.task.taskId;
|
|
expect(createResult.task.status).toBe('working');
|
|
|
|
// Phase 2: Wait for server to queue elicitation and update status
|
|
const task = await waitForTaskStatus(
|
|
id =>
|
|
elicitClient.request(
|
|
{
|
|
method: 'tasks/get',
|
|
params: { taskId: id }
|
|
},
|
|
TaskSchema
|
|
),
|
|
taskId,
|
|
'input_required',
|
|
{
|
|
intervalMs: createResult.task.pollInterval ?? 100
|
|
}
|
|
);
|
|
|
|
// Verify we saw input_required status (not completed or failed)
|
|
expect(task.status).toBe('input_required');
|
|
|
|
// Phase 3: Call tasks/result to dequeue messages and get final result
|
|
// This should:
|
|
// - Deliver the queued elicitation request via SSE
|
|
// - Client handler responds
|
|
// - Server receives response, completes task
|
|
// - Return final result
|
|
const result = await elicitClient.request(
|
|
{
|
|
method: 'tasks/result',
|
|
params: { taskId }
|
|
},
|
|
CallToolResultSchema
|
|
);
|
|
|
|
// Verify elicitation was received and processed
|
|
expect(elicitationReceived).toBe(true);
|
|
|
|
// Verify the elicitation request had related-task metadata
|
|
expect(elicitationRequestMeta).toBeDefined();
|
|
expect(elicitationRequestMeta?.[RELATED_TASK_META_KEY]).toEqual({ taskId });
|
|
|
|
// Verify final result
|
|
expect(result.content).toEqual([{ type: 'text', text: 'Hello, TestUser!' }]);
|
|
|
|
// Verify task is now completed
|
|
const finalTask = await elicitClient.request(
|
|
{
|
|
method: 'tasks/get',
|
|
params: { taskId }
|
|
},
|
|
TaskSchema
|
|
);
|
|
expect(finalTask.status).toBe('completed');
|
|
|
|
await transport.close();
|
|
}, 15000);
|
|
});
|
|
|
|
describe('Task Listing and Pagination', () => {
|
|
it('should list tasks', async () => {
|
|
const client = new Client({
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Create multiple tasks
|
|
const taskIds: string[] = [];
|
|
for (let i = 0; i < 3; i++) {
|
|
const createResult = await client.request(
|
|
{
|
|
method: 'tools/call',
|
|
params: {
|
|
name: 'long-task',
|
|
arguments: {
|
|
duration: 1000
|
|
},
|
|
task: {
|
|
ttl: 60000
|
|
}
|
|
}
|
|
},
|
|
CreateTaskResultSchema
|
|
);
|
|
taskIds.push(createResult.task.taskId);
|
|
}
|
|
|
|
// List tasks using taskStore
|
|
const listResult = await taskStore.listTasks();
|
|
|
|
expect(listResult.tasks.length).toBeGreaterThanOrEqual(3);
|
|
expect(listResult.tasks.some(t => taskIds.includes(t.taskId))).toBe(true);
|
|
|
|
await transport.close();
|
|
});
|
|
|
|
it('should handle pagination with large datasets', async () => {
|
|
const client = new Client({
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Create 15 tasks (more than page size of 10)
|
|
for (let i = 0; i < 15; i++) {
|
|
await client.request(
|
|
{
|
|
method: 'tools/call',
|
|
params: {
|
|
name: 'long-task',
|
|
arguments: {
|
|
duration: 5000
|
|
},
|
|
task: {
|
|
ttl: 60000
|
|
}
|
|
}
|
|
},
|
|
CreateTaskResultSchema
|
|
);
|
|
}
|
|
|
|
// Get first page using taskStore
|
|
const page1 = await taskStore.listTasks();
|
|
|
|
expect(page1.tasks.length).toBe(10);
|
|
expect(page1.nextCursor).toBeDefined();
|
|
|
|
// Get second page
|
|
const page2 = await taskStore.listTasks(page1.nextCursor);
|
|
|
|
expect(page2.tasks.length).toBeGreaterThanOrEqual(5);
|
|
|
|
await transport.close();
|
|
});
|
|
});
|
|
|
|
describe('Error Handling', () => {
|
|
it('should return error code -32602 for non-existent task in tasks/get', async () => {
|
|
const client = new Client({
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Try to get non-existent task via tasks/get request
|
|
await expect(client.experimental.tasks.getTask('non-existent-task-id')).rejects.toSatisfy((error: McpError) => {
|
|
expect(error).toBeInstanceOf(McpError);
|
|
expect(error.code).toBe(ErrorCode.InvalidParams);
|
|
expect(error.message).toContain('Task not found');
|
|
return true;
|
|
});
|
|
|
|
await transport.close();
|
|
});
|
|
|
|
it('should return error code -32602 for non-existent task in tasks/cancel', async () => {
|
|
const client = new Client({
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Try to cancel non-existent task via tasks/cancel request
|
|
await expect(client.experimental.tasks.cancelTask('non-existent-task-id')).rejects.toSatisfy((error: McpError) => {
|
|
expect(error).toBeInstanceOf(McpError);
|
|
expect(error.code).toBe(ErrorCode.InvalidParams);
|
|
expect(error.message).toContain('Task not found');
|
|
return true;
|
|
});
|
|
|
|
await transport.close();
|
|
});
|
|
|
|
it('should return error code -32602 for non-existent task in tasks/result', async () => {
|
|
const client = new Client({
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Try to get result of non-existent task via tasks/result request
|
|
await expect(
|
|
client.request(
|
|
{
|
|
method: 'tasks/result',
|
|
params: { taskId: 'non-existent-task-id' }
|
|
},
|
|
CallToolResultSchema
|
|
)
|
|
).rejects.toSatisfy((error: McpError) => {
|
|
expect(error).toBeInstanceOf(McpError);
|
|
expect(error.code).toBe(ErrorCode.InvalidParams);
|
|
expect(error.message).toContain('Task not found');
|
|
return true;
|
|
});
|
|
|
|
await transport.close();
|
|
});
|
|
});
|
|
|
|
describe('TTL and Cleanup', () => {
|
|
it('should respect TTL in task creation', async () => {
|
|
const client = new Client({
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Create a task with specific TTL
|
|
const createResult = await client.request(
|
|
{
|
|
method: 'tools/call',
|
|
params: {
|
|
name: 'long-task',
|
|
arguments: {
|
|
duration: 100
|
|
},
|
|
task: {
|
|
ttl: 5000
|
|
}
|
|
}
|
|
},
|
|
CreateTaskResultSchema
|
|
);
|
|
|
|
const taskId = createResult.task.taskId;
|
|
|
|
// Verify TTL is set correctly
|
|
expect(createResult.task.ttl).toBe(60000); // The task store uses 60000 as default
|
|
|
|
// Task should exist
|
|
const task = await client.request(
|
|
{
|
|
method: 'tasks/get',
|
|
params: { taskId }
|
|
},
|
|
TaskSchema
|
|
);
|
|
expect(task).toBeDefined();
|
|
expect(task.ttl).toBe(60000);
|
|
|
|
await transport.close();
|
|
});
|
|
});
|
|
|
|
describe('Task Cancellation with Queued Messages', () => {
|
|
it('should clear queue and deliver no messages when task is cancelled before tasks/result', async () => {
|
|
// Register a tool that queues messages but doesn't complete immediately
|
|
mcpServer.experimental.tasks.registerToolTask(
|
|
'cancellable-task',
|
|
{
|
|
title: 'Cancellable Task',
|
|
description: 'A tool that queues messages and can be cancelled',
|
|
inputSchema: {
|
|
messageCount: z.number().describe('Number of messages to queue').default(2)
|
|
}
|
|
},
|
|
{
|
|
async createTask({ messageCount }, extra) {
|
|
const task = await extra.taskStore.createTask({
|
|
ttl: 60000,
|
|
pollInterval: 100
|
|
});
|
|
|
|
// Perform async work that queues messages
|
|
(async () => {
|
|
try {
|
|
await new Promise(resolve => setTimeout(resolve, 100));
|
|
|
|
// Queue multiple elicitation requests
|
|
for (let i = 0; i < messageCount; i++) {
|
|
// Send request but don't await - let it queue
|
|
extra
|
|
.sendRequest(
|
|
{
|
|
method: 'elicitation/create',
|
|
params: {
|
|
mode: 'form',
|
|
message: `Message ${i + 1} of ${messageCount}`,
|
|
requestedSchema: {
|
|
type: 'object',
|
|
properties: {
|
|
response: { type: 'string' }
|
|
},
|
|
required: ['response']
|
|
}
|
|
}
|
|
},
|
|
ElicitResultSchema,
|
|
{ relatedTask: { taskId: task.taskId } } as unknown as TaskRequestOptions
|
|
)
|
|
.catch(() => {
|
|
// Ignore errors from cancelled requests
|
|
});
|
|
}
|
|
|
|
// Don't complete - let the task be cancelled
|
|
// Wait indefinitely (or until cancelled)
|
|
await new Promise(() => {});
|
|
} catch {
|
|
// Ignore errors - task was cancelled
|
|
}
|
|
})().catch(() => {
|
|
// Catch any unhandled errors from the async execution
|
|
});
|
|
|
|
return { task };
|
|
},
|
|
async getTask(_args, extra) {
|
|
const task = await extra.taskStore.getTask(extra.taskId);
|
|
if (!task) {
|
|
throw new Error(`Task ${extra.taskId} not found`);
|
|
}
|
|
return task;
|
|
},
|
|
async getTaskResult(_args, extra) {
|
|
const result = await extra.taskStore.getTaskResult(extra.taskId);
|
|
return result as { content: Array<{ type: 'text'; text: string }> };
|
|
}
|
|
}
|
|
);
|
|
|
|
const client = new Client(
|
|
{
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
},
|
|
{
|
|
capabilities: {
|
|
elicitation: {}
|
|
}
|
|
}
|
|
);
|
|
|
|
let elicitationCallCount = 0;
|
|
|
|
// Set up elicitation handler to track if any messages are delivered
|
|
client.setRequestHandler(ElicitRequestSchema, async () => {
|
|
elicitationCallCount++;
|
|
return {
|
|
action: 'accept' as const,
|
|
content: {
|
|
response: 'Should not be called'
|
|
}
|
|
};
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Create a task that will queue messages
|
|
const createResult = await client.request(
|
|
{
|
|
method: 'tools/call',
|
|
params: {
|
|
name: 'cancellable-task',
|
|
arguments: {
|
|
messageCount: 2
|
|
},
|
|
task: {
|
|
ttl: 60000
|
|
}
|
|
}
|
|
},
|
|
CreateTaskResultSchema
|
|
);
|
|
|
|
const taskId = createResult.task.taskId;
|
|
|
|
// Wait for messages to be queued
|
|
await new Promise(resolve => setTimeout(resolve, 200));
|
|
|
|
// Verify task is in input_required state and messages are queued
|
|
let task = await client.request(
|
|
{
|
|
method: 'tasks/get',
|
|
params: { taskId }
|
|
},
|
|
TaskSchema
|
|
);
|
|
expect(task.status).toBe('input_required');
|
|
|
|
// Cancel the task before calling tasks/result using the proper tasks/cancel request
|
|
// This will trigger queue cleanup via _clearTaskQueue in the handler
|
|
await client.request(
|
|
{
|
|
method: 'tasks/cancel',
|
|
params: { taskId }
|
|
},
|
|
z.object({ _meta: z.record(z.unknown()).optional() })
|
|
);
|
|
|
|
// Verify task is cancelled
|
|
task = await client.request(
|
|
{
|
|
method: 'tasks/get',
|
|
params: { taskId }
|
|
},
|
|
TaskSchema
|
|
);
|
|
expect(task.status).toBe('cancelled');
|
|
|
|
// Attempt to call tasks/result
|
|
// When a task is cancelled, the system needs to clear the message queue
|
|
// and reject any pending message delivery promises, meaning no further
|
|
// messages should be delivered for a cancelled task.
|
|
try {
|
|
await client.request(
|
|
{
|
|
method: 'tasks/result',
|
|
params: { taskId }
|
|
},
|
|
CallToolResultSchema
|
|
);
|
|
} catch {
|
|
// tasks/result might throw an error for cancelled tasks without a result
|
|
// This is acceptable behavior
|
|
}
|
|
|
|
// Verify no elicitation messages were delivered, as the queue should be cleared immediately on cancellation
|
|
expect(elicitationCallCount).toBe(0);
|
|
|
|
// Verify queue remains cleared on subsequent calls
|
|
try {
|
|
await client.request(
|
|
{
|
|
method: 'tasks/result',
|
|
params: { taskId }
|
|
},
|
|
CallToolResultSchema
|
|
);
|
|
} catch {
|
|
// Expected - task is cancelled
|
|
}
|
|
|
|
// Still no messages should have been delivered
|
|
expect(elicitationCallCount).toBe(0);
|
|
|
|
await transport.close();
|
|
}, 10000);
|
|
});
|
|
|
|
describe('Continuous Message Delivery', () => {
|
|
it('should deliver messages immediately while tasks/result is blocking', async () => {
|
|
// Register a tool that queues messages over time
|
|
mcpServer.experimental.tasks.registerToolTask(
|
|
'streaming-task',
|
|
{
|
|
title: 'Streaming Task',
|
|
description: 'A tool that sends messages over time',
|
|
inputSchema: {
|
|
messageCount: z.number().describe('Number of messages to send').default(3),
|
|
delayBetweenMessages: z.number().describe('Delay between messages in ms').default(200)
|
|
}
|
|
},
|
|
{
|
|
async createTask({ messageCount, delayBetweenMessages }, extra) {
|
|
const task = await extra.taskStore.createTask({
|
|
ttl: 60000,
|
|
pollInterval: 100
|
|
});
|
|
|
|
// Perform async work that sends messages over time
|
|
(async () => {
|
|
try {
|
|
// Wait a bit before starting to send messages
|
|
await new Promise(resolve => setTimeout(resolve, 100));
|
|
|
|
const responses: string[] = [];
|
|
|
|
// Send messages with delays between them
|
|
for (let i = 0; i < messageCount; i++) {
|
|
const elicitationResult = await extra.sendRequest(
|
|
{
|
|
method: 'elicitation/create',
|
|
params: {
|
|
mode: 'form',
|
|
message: `Streaming message ${i + 1} of ${messageCount}`,
|
|
requestedSchema: {
|
|
type: 'object',
|
|
properties: {
|
|
response: { type: 'string' }
|
|
},
|
|
required: ['response']
|
|
}
|
|
}
|
|
},
|
|
ElicitResultSchema,
|
|
{ relatedTask: { taskId: task.taskId } } as unknown as TaskRequestOptions
|
|
);
|
|
|
|
if (elicitationResult.action === 'accept' && elicitationResult.content) {
|
|
responses.push(elicitationResult.content.response as string);
|
|
}
|
|
|
|
// Wait before sending next message (if not the last one)
|
|
if (i < messageCount - 1) {
|
|
await new Promise(resolve => setTimeout(resolve, delayBetweenMessages));
|
|
}
|
|
}
|
|
|
|
// Complete with all responses
|
|
try {
|
|
await extra.taskStore.storeTaskResult(task.taskId, 'completed', {
|
|
content: [{ type: 'text', text: `Received all responses: ${responses.join(', ')}` }]
|
|
});
|
|
} catch {
|
|
// Task may have been cleaned up if test ended
|
|
}
|
|
} catch (error) {
|
|
// Handle errors
|
|
try {
|
|
await extra.taskStore.storeTaskResult(task.taskId, 'failed', {
|
|
content: [{ type: 'text', text: `Error: ${error}` }],
|
|
isError: true
|
|
});
|
|
} catch {
|
|
// Task may have been cleaned up if test ended
|
|
}
|
|
}
|
|
})();
|
|
|
|
return { task };
|
|
},
|
|
async getTask(_args, extra) {
|
|
const task = await extra.taskStore.getTask(extra.taskId);
|
|
if (!task) {
|
|
throw new Error(`Task ${extra.taskId} not found`);
|
|
}
|
|
return task;
|
|
},
|
|
async getTaskResult(_args, extra) {
|
|
const result = await extra.taskStore.getTaskResult(extra.taskId);
|
|
return result as { content: Array<{ type: 'text'; text: string }> };
|
|
}
|
|
}
|
|
);
|
|
|
|
const client = new Client(
|
|
{
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
},
|
|
{
|
|
capabilities: {
|
|
elicitation: {}
|
|
}
|
|
}
|
|
);
|
|
|
|
const receivedMessages: Array<{ message: string; timestamp: number }> = [];
|
|
let tasksResultStartTime = 0;
|
|
|
|
// Set up elicitation handler to track when messages arrive
|
|
client.setRequestHandler(ElicitRequestSchema, async request => {
|
|
const timestamp = Date.now();
|
|
receivedMessages.push({
|
|
message: request.params.message,
|
|
timestamp
|
|
});
|
|
|
|
// Extract the message number
|
|
const match = request.params.message.match(/Streaming message (\d+) of (\d+)/);
|
|
const messageNum = match ? match[1] : 'unknown';
|
|
|
|
// Respond immediately
|
|
return {
|
|
action: 'accept' as const,
|
|
content: {
|
|
response: `Response ${messageNum}`
|
|
}
|
|
};
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Create a task that will send messages over time
|
|
const createResult = await client.request(
|
|
{
|
|
method: 'tools/call',
|
|
params: {
|
|
name: 'streaming-task',
|
|
arguments: {
|
|
messageCount: 3,
|
|
delayBetweenMessages: 300
|
|
},
|
|
task: {
|
|
ttl: 60000
|
|
}
|
|
}
|
|
},
|
|
CreateTaskResultSchema
|
|
);
|
|
|
|
const taskId = createResult.task.taskId;
|
|
|
|
// Verify task is in working status
|
|
let task = await client.request(
|
|
{
|
|
method: 'tasks/get',
|
|
params: { taskId }
|
|
},
|
|
TaskSchema
|
|
);
|
|
expect(task.status).toBe('working');
|
|
|
|
// Call tasks/result immediately (before messages are queued)
|
|
// This should block and deliver messages as they arrive
|
|
tasksResultStartTime = Date.now();
|
|
const resultPromise = client.request(
|
|
{
|
|
method: 'tasks/result',
|
|
params: { taskId }
|
|
},
|
|
CallToolResultSchema
|
|
);
|
|
|
|
// Wait for the task to complete and get the result
|
|
const result = await resultPromise;
|
|
|
|
// Verify all 3 messages were delivered
|
|
expect(receivedMessages.length).toBe(3);
|
|
expect(receivedMessages[0].message).toBe('Streaming message 1 of 3');
|
|
expect(receivedMessages[1].message).toBe('Streaming message 2 of 3');
|
|
expect(receivedMessages[2].message).toBe('Streaming message 3 of 3');
|
|
|
|
// Verify messages were delivered over time (not all at once)
|
|
// The delay between messages should be approximately 300ms
|
|
const timeBetweenFirstAndSecond = receivedMessages[1].timestamp - receivedMessages[0].timestamp;
|
|
const timeBetweenSecondAndThird = receivedMessages[2].timestamp - receivedMessages[1].timestamp;
|
|
|
|
// Allow some tolerance for timing (messages should be at least 200ms apart)
|
|
expect(timeBetweenFirstAndSecond).toBeGreaterThan(200);
|
|
expect(timeBetweenSecondAndThird).toBeGreaterThan(200);
|
|
|
|
// Verify messages were delivered while tasks/result was blocking
|
|
// (all messages should arrive after tasks/result was called)
|
|
for (const msg of receivedMessages) {
|
|
expect(msg.timestamp).toBeGreaterThanOrEqual(tasksResultStartTime);
|
|
}
|
|
|
|
// Verify final result is correct
|
|
expect(result.content).toEqual([{ type: 'text', text: 'Received all responses: Response 1, Response 2, Response 3' }]);
|
|
|
|
// Verify task is now completed
|
|
task = await client.request(
|
|
{
|
|
method: 'tasks/get',
|
|
params: { taskId }
|
|
},
|
|
TaskSchema
|
|
);
|
|
expect(task.status).toBe('completed');
|
|
|
|
await transport.close();
|
|
}, 15000); // Increase timeout to 15 seconds to allow for message delays
|
|
});
|
|
|
|
describe('Terminal Task with Queued Messages', () => {
|
|
it('should deliver queued messages followed by final result for terminal task', async () => {
|
|
// Register a tool that completes quickly and queues messages before completion
|
|
mcpServer.experimental.tasks.registerToolTask(
|
|
'quick-complete-task',
|
|
{
|
|
title: 'Quick Complete Task',
|
|
description: 'A tool that queues messages and completes quickly',
|
|
inputSchema: {
|
|
messageCount: z.number().describe('Number of messages to queue').default(2)
|
|
}
|
|
},
|
|
{
|
|
async createTask({ messageCount }, extra) {
|
|
const task = await extra.taskStore.createTask({
|
|
ttl: 60000,
|
|
pollInterval: 100
|
|
});
|
|
|
|
// Perform async work that queues messages and completes quickly
|
|
(async () => {
|
|
try {
|
|
// Queue messages - these will be queued before the task completes
|
|
// We await each one starting to ensure they're queued before completing
|
|
for (let i = 0; i < messageCount; i++) {
|
|
// Start the request but don't wait for response
|
|
// The request gets queued when sendRequest is called
|
|
extra
|
|
.sendRequest(
|
|
{
|
|
method: 'elicitation/create',
|
|
params: {
|
|
mode: 'form',
|
|
message: `Quick message ${i + 1} of ${messageCount}`,
|
|
requestedSchema: {
|
|
type: 'object',
|
|
properties: {
|
|
response: { type: 'string' }
|
|
},
|
|
required: ['response']
|
|
}
|
|
}
|
|
},
|
|
ElicitResultSchema,
|
|
{ relatedTask: { taskId: task.taskId } } as unknown as TaskRequestOptions
|
|
)
|
|
.catch(() => {});
|
|
// Small delay to ensure message is queued before next iteration
|
|
await new Promise(resolve => setTimeout(resolve, 10));
|
|
}
|
|
|
|
// Complete the task after all messages are queued
|
|
try {
|
|
await extra.taskStore.storeTaskResult(task.taskId, 'completed', {
|
|
content: [{ type: 'text', text: 'Task completed quickly' }]
|
|
});
|
|
} catch {
|
|
// Task may have been cleaned up if test ended
|
|
}
|
|
} catch (error) {
|
|
// Handle errors
|
|
try {
|
|
await extra.taskStore.storeTaskResult(task.taskId, 'failed', {
|
|
content: [{ type: 'text', text: `Error: ${error}` }],
|
|
isError: true
|
|
});
|
|
} catch {
|
|
// Task may have been cleaned up if test ended
|
|
}
|
|
}
|
|
})();
|
|
|
|
return { task };
|
|
},
|
|
async getTask(_args, extra) {
|
|
const task = await extra.taskStore.getTask(extra.taskId);
|
|
if (!task) {
|
|
throw new Error(`Task ${extra.taskId} not found`);
|
|
}
|
|
return task;
|
|
},
|
|
async getTaskResult(_args, extra) {
|
|
const result = await extra.taskStore.getTaskResult(extra.taskId);
|
|
return result as { content: Array<{ type: 'text'; text: string }> };
|
|
}
|
|
}
|
|
);
|
|
|
|
const client = new Client(
|
|
{
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
},
|
|
{
|
|
capabilities: {
|
|
elicitation: {}
|
|
}
|
|
}
|
|
);
|
|
|
|
const receivedMessages: Array<{ type: string; message?: string; content?: unknown }> = [];
|
|
|
|
// Set up elicitation handler to track message order
|
|
client.setRequestHandler(ElicitRequestSchema, async request => {
|
|
receivedMessages.push({
|
|
type: 'elicitation',
|
|
message: request.params.message
|
|
});
|
|
|
|
// Extract the message number
|
|
const match = request.params.message.match(/Quick message (\d+) of (\d+)/);
|
|
const messageNum = match ? match[1] : 'unknown';
|
|
|
|
return {
|
|
action: 'accept' as const,
|
|
content: {
|
|
response: `Response ${messageNum}`
|
|
}
|
|
};
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Create a task that will complete quickly with queued messages
|
|
const createResult = await client.request(
|
|
{
|
|
method: 'tools/call',
|
|
params: {
|
|
name: 'quick-complete-task',
|
|
arguments: {
|
|
messageCount: 2
|
|
},
|
|
task: {
|
|
ttl: 60000
|
|
}
|
|
}
|
|
},
|
|
CreateTaskResultSchema
|
|
);
|
|
|
|
const taskId = createResult.task.taskId;
|
|
|
|
// Wait for task to complete and messages to be queued
|
|
const task = await waitForTaskStatus(id => taskStore.getTask(id), taskId, 'completed');
|
|
|
|
// Verify task is in terminal status (completed)
|
|
expect(task.status).toBe('completed');
|
|
|
|
// Call tasks/result - should deliver queued messages followed by final result
|
|
const result = await client.request(
|
|
{
|
|
method: 'tasks/result',
|
|
params: { taskId }
|
|
},
|
|
CallToolResultSchema
|
|
);
|
|
|
|
// Verify all queued messages were delivered before the final result
|
|
expect(receivedMessages.length).toBe(2);
|
|
expect(receivedMessages[0].message).toBe('Quick message 1 of 2');
|
|
expect(receivedMessages[1].message).toBe('Quick message 2 of 2');
|
|
|
|
// Verify final result is correct
|
|
expect(result.content).toEqual([{ type: 'text', text: 'Task completed quickly' }]);
|
|
|
|
// Verify queue is cleaned up - calling tasks/result again should only return the result
|
|
receivedMessages.length = 0; // Clear the array
|
|
|
|
const result2 = await client.request(
|
|
{
|
|
method: 'tasks/result',
|
|
params: { taskId }
|
|
},
|
|
CallToolResultSchema
|
|
);
|
|
|
|
// No messages should be delivered on second call (queue was cleaned up)
|
|
expect(receivedMessages.length).toBe(0);
|
|
expect(result2.content).toEqual([{ type: 'text', text: 'Task completed quickly' }]);
|
|
|
|
await transport.close();
|
|
}, 10000);
|
|
});
|
|
|
|
describe('Concurrent Operations', () => {
|
|
it('should handle multiple concurrent task creations', async () => {
|
|
const client = new Client({
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Create multiple tasks concurrently
|
|
const promises = Array.from({ length: 5 }, () =>
|
|
client.request(
|
|
{
|
|
method: 'tools/call',
|
|
params: {
|
|
name: 'long-task',
|
|
arguments: {
|
|
duration: 500
|
|
},
|
|
task: {
|
|
ttl: 60000
|
|
}
|
|
}
|
|
},
|
|
CreateTaskResultSchema
|
|
)
|
|
);
|
|
|
|
const results = await Promise.all(promises);
|
|
|
|
// Verify all tasks were created with unique IDs
|
|
const taskIds = results.map(r => r.task.taskId);
|
|
expect(new Set(taskIds).size).toBe(5);
|
|
|
|
// Verify all tasks are in working status
|
|
for (const result of results) {
|
|
expect(result.task.status).toBe('working');
|
|
}
|
|
|
|
await transport.close();
|
|
});
|
|
|
|
it('should handle concurrent operations on same task', async () => {
|
|
const client = new Client({
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Create a task
|
|
const createResult = await client.request(
|
|
{
|
|
method: 'tools/call',
|
|
params: {
|
|
name: 'long-task',
|
|
arguments: {
|
|
duration: 2000
|
|
},
|
|
task: {
|
|
ttl: 60000
|
|
}
|
|
}
|
|
},
|
|
CreateTaskResultSchema
|
|
);
|
|
|
|
const taskId = createResult.task.taskId;
|
|
|
|
// Perform multiple concurrent gets
|
|
const getPromises = Array.from({ length: 5 }, () =>
|
|
client.request(
|
|
{
|
|
method: 'tasks/get',
|
|
params: { taskId }
|
|
},
|
|
TaskSchema
|
|
)
|
|
);
|
|
|
|
const tasks = await Promise.all(getPromises);
|
|
|
|
// All should return the same task
|
|
for (const task of tasks) {
|
|
expect(task.taskId).toBe(taskId);
|
|
expect(task.status).toBe('working');
|
|
}
|
|
|
|
await transport.close();
|
|
});
|
|
});
|
|
|
|
describe('callToolStream with elicitation', () => {
|
|
it('should deliver elicitation via callToolStream and complete task', async () => {
|
|
const client = new Client(
|
|
{
|
|
name: 'test-client',
|
|
version: '1.0.0'
|
|
},
|
|
{
|
|
capabilities: {
|
|
elicitation: {}
|
|
}
|
|
}
|
|
);
|
|
|
|
// Track elicitation request receipt
|
|
let elicitationReceived = false;
|
|
let elicitationMessage = '';
|
|
|
|
// Set up elicitation handler on client
|
|
client.setRequestHandler(ElicitRequestSchema, async request => {
|
|
elicitationReceived = true;
|
|
elicitationMessage = request.params.message;
|
|
|
|
return {
|
|
action: 'accept' as const,
|
|
content: {
|
|
userName: 'StreamUser'
|
|
}
|
|
};
|
|
});
|
|
|
|
const transport = new StreamableHTTPClientTransport(baseUrl);
|
|
await client.connect(transport);
|
|
|
|
// Use callToolStream instead of raw request()
|
|
const stream = client.experimental.tasks.callToolStream({ name: 'input-task', arguments: {} }, CallToolResultSchema, {
|
|
task: { ttl: 60000 }
|
|
});
|
|
|
|
// Collect all stream messages
|
|
const messages: Array<{ type: string; task?: unknown; result?: unknown; error?: unknown }> = [];
|
|
for await (const message of stream) {
|
|
messages.push(message);
|
|
}
|
|
|
|
// Verify stream yielded expected message types
|
|
expect(messages.length).toBeGreaterThanOrEqual(2);
|
|
|
|
// First message should be taskCreated
|
|
expect(messages[0].type).toBe('taskCreated');
|
|
expect(messages[0].task).toBeDefined();
|
|
|
|
// Should have a taskStatus message
|
|
const statusMessages = messages.filter(m => m.type === 'taskStatus');
|
|
expect(statusMessages.length).toBeGreaterThanOrEqual(1);
|
|
|
|
// Last message should be result
|
|
const lastMessage = messages[messages.length - 1];
|
|
expect(lastMessage.type).toBe('result');
|
|
expect(lastMessage.result).toBeDefined();
|
|
|
|
// Verify elicitation was received and processed
|
|
expect(elicitationReceived).toBe(true);
|
|
expect(elicitationMessage).toContain('What is your name?');
|
|
|
|
// Verify result content
|
|
const result = lastMessage.result as { content: Array<{ type: string; text: string }> };
|
|
expect(result.content).toEqual([{ type: 'text', text: 'Hello, StreamUser!' }]);
|
|
|
|
await transport.close();
|
|
}, 15000);
|
|
});
|
|
});
|