diff --git a/packages/core/src/mcp/client.ts b/packages/core/src/mcp/client.ts index 0b10ecd7e4..5f48cba649 100644 --- a/packages/core/src/mcp/client.ts +++ b/packages/core/src/mcp/client.ts @@ -3,7 +3,7 @@ export * as MCPClient from "./client" import path from "node:path" import { execFile } from "node:child_process" import { pathToFileURL } from "node:url" -import { Client, type ClientOptions } from "@modelcontextprotocol/sdk/client/index.js" +import { Client } from "@modelcontextprotocol/sdk/client/index.js" import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js" import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js" import { UnauthorizedError, type OAuthClientProvider } from "@modelcontextprotocol/sdk/client/auth.js" @@ -21,6 +21,7 @@ import { ListToolsResultSchema, PromptListChangedNotificationSchema, PromptSchema, + ResourceListChangedNotificationSchema, type LoggingMessageNotification, LoggingMessageNotificationSchema, ToolListChangedNotificationSchema, @@ -68,11 +69,13 @@ export interface ToolDefinition { export interface PromptDefinition { readonly name: string readonly description: string | undefined - readonly arguments: ReadonlyArray<{ - readonly name: string - readonly description: string | undefined - readonly required: boolean | undefined - }> | undefined + readonly arguments: + | ReadonlyArray<{ + readonly name: string + readonly description: string | undefined + readonly required: boolean | undefined + }> + | undefined } export interface PromptMessage { @@ -84,6 +87,28 @@ export interface PromptResult { readonly messages: ReadonlyArray } +export interface ResourceDefinition { + readonly name: string + readonly uri: string + readonly description: string | undefined + readonly mimeType: string | undefined +} + +export interface ResourceTemplateDefinition { + readonly name: string + readonly uriTemplate: string + readonly description: string | undefined + readonly mimeType: string | undefined +} + +export type ResourceContentPart = + | { readonly type: "text"; readonly uri: string; readonly text: string; readonly mimeType: string | undefined } + | { readonly type: "blob"; readonly uri: string; readonly blob: string; readonly mimeType: string | undefined } + +export interface ReadResourceResult { + readonly contents: ReadonlyArray +} + export type CallToolContent = | { readonly type: "text"; readonly text: string } | { readonly type: "media"; readonly data: string; readonly mimeType: string } @@ -124,6 +149,12 @@ export interface Connection { readonly tools: () => Effect.Effect /** Lists the server's prompts; returns [] when the server doesn't advertise prompt support, fails on a transport error. */ readonly prompts: () => Effect.Effect + /** Lists the server's resources; returns [] when the server doesn't advertise resource support. */ + readonly resources: () => Effect.Effect + /** Lists the server's resource templates; returns [] when the server doesn't advertise resource support. */ + readonly resourceTemplates: () => Effect.Effect + /** Reads one resource; returns undefined when the server doesn't advertise resource support. */ + readonly readResource: (input: { readonly uri: string }) => Effect.Effect /** Invokes a prompt on the server. Interruption aborts the in-flight request. */ readonly prompt: (input: { readonly name: string @@ -141,6 +172,8 @@ export interface Connection { readonly onToolsChanged: (callback: () => void) => void /** Registers a callback fired when the server announces its prompt list changed; no-op if unsupported. */ readonly onPromptsChanged: (callback: () => void) => void + /** Registers a callback fired when the server announces its resource catalog changed. */ + readonly onResourcesChanged: (callback: () => void) => void } /** Connects an MCP server; closing the calling scope tears down the transport and any spawned process. */ @@ -168,7 +201,8 @@ export const connect = Effect.fnUntraced(function* ( }, }) } - if (!URL.canParse(config.url)) return yield* new ConnectError({ server, message: `Invalid MCP URL for "${server}"` }) + if (!URL.canParse(config.url)) + return yield* new ConnectError({ server, message: `Invalid MCP URL for "${server}"` }) return new StreamableHTTPClientTransport(new URL(config.url), { requestInit: config.headers ? { headers: config.headers } : undefined, authProvider, @@ -202,10 +236,7 @@ export const connect = Effect.fnUntraced(function* ( }).pipe(Effect.exit) if (Exit.isSuccess(exit)) { yield* Effect.addFinalizer(() => - cleanupStdioDescendants(transport).pipe( - Effect.andThen(Effect.promise(() => client.close())), - Effect.ignore, - ), + cleanupStdioDescendants(transport).pipe(Effect.andThen(Effect.promise(() => client.close())), Effect.ignore), ) const catalogTimeout = config.timeout?.catalog ?? DEFAULT_CATALOG_TIMEOUT const executionTimeout = config.timeout?.execution ?? DEFAULT_EXECUTION_TIMEOUT @@ -257,7 +288,9 @@ export const connect = Effect.fnUntraced(function* ( ), catch: (error) => (error instanceof Error ? error : new Error(String(error))), }).pipe( - Effect.tapError((error) => Effect.logWarning("failed to list MCP prompts", { server, error: error.message })), + Effect.tapError((error) => + Effect.logWarning("failed to list MCP prompts", { server, error: error.message }), + ), ) return prompts.map((prompt) => ({ name: prompt.name, @@ -269,6 +302,74 @@ export const connect = Effect.fnUntraced(function* ( })), })) }), + resources: () => + Effect.gen(function* () { + if (!client.getServerCapabilities()?.resources) return [] + const resources = yield* Effect.tryPromise({ + try: () => + paginate( + (cursor) => + client.listResources(cursor === undefined ? undefined : { cursor }, { timeout: catalogTimeout }), + (result) => result.resources, + ), + catch: (error) => (error instanceof Error ? error : new Error(String(error))), + }).pipe( + Effect.tapError((error) => + Effect.logWarning("failed to list MCP resources", { server, error: error.message }), + ), + ) + return resources.map((resource) => ({ + name: resource.name, + uri: resource.uri, + description: resource.description, + mimeType: resource.mimeType, + })) + }), + resourceTemplates: () => + Effect.gen(function* () { + if (!client.getServerCapabilities()?.resources) return [] + const templates = yield* Effect.tryPromise({ + try: () => + paginate( + (cursor) => + client.listResourceTemplates(cursor === undefined ? undefined : { cursor }, { + timeout: catalogTimeout, + }), + (result) => result.resourceTemplates, + ), + catch: (error) => (error instanceof Error ? error : new Error(String(error))), + }).pipe( + Effect.tapError((error) => + Effect.logWarning("failed to list MCP resource templates", { server, error: error.message }), + ), + ) + return templates.map((template) => ({ + name: template.name, + uriTemplate: template.uriTemplate, + description: template.description, + mimeType: template.mimeType, + })) + }), + readResource: (input) => + Effect.gen(function* () { + if (!client.getServerCapabilities()?.resources) return undefined + const result = yield* Effect.tryPromise({ + try: (signal) => client.readResource({ uri: input.uri }, { signal, timeout: executionTimeout }), + catch: (error) => (error instanceof Error ? error : new Error(String(error))), + }).pipe( + Effect.tapError((error) => + Effect.logWarning("failed to read MCP resource", { server, uri: input.uri, error: error.message }), + ), + ) + return { + contents: result.contents.map( + (part): ResourceContentPart => + "text" in part + ? { type: "text", uri: part.uri, text: part.text, mimeType: part.mimeType } + : { type: "blob", uri: part.uri, blob: part.blob, mimeType: part.mimeType }, + ), + } + }), prompt: (input) => Effect.tryPromise({ try: (signal) => @@ -328,13 +429,14 @@ export const connect = Effect.fnUntraced(function* ( if (!client.getServerCapabilities()?.prompts?.listChanged) return client.setNotificationHandler(PromptListChangedNotificationSchema, async () => callback()) }, + onResourcesChanged: (callback) => { + if (!client.getServerCapabilities()?.resources?.listChanged) return + client.setNotificationHandler(ResourceListChangedNotificationSchema, async () => callback()) + }, } satisfies Connection } - yield* cleanupStdioDescendants(transport).pipe( - Effect.andThen(Effect.promise(() => transport.close())), - Effect.ignore, - ) + yield* cleanupStdioDescendants(transport).pipe(Effect.andThen(Effect.promise(() => transport.close())), Effect.ignore) const error = Cause.squash(exit.cause) if (error instanceof UnauthorizedError) return yield* new NeedsAuthError({ server }) return yield* new ConnectError({ server, message: error instanceof Error ? error.message : String(error) }) diff --git a/packages/core/src/mcp/index.ts b/packages/core/src/mcp/index.ts index 1ce8c3d3ec..7a99c15982 100644 --- a/packages/core/src/mcp/index.ts +++ b/packages/core/src/mcp/index.ts @@ -83,48 +83,16 @@ export class PromptResult extends Schema.Class("MCP.PromptResult") messages: Schema.Array(PromptMessage), }) {} -export class Resource extends Schema.Class("MCP.Resource")({ - server: ServerName, - name: Schema.String, - uri: Schema.String, - description: Schema.String.pipe(Schema.optional), - mimeType: Schema.String.pipe(Schema.optional), -}) {} - -export class ResourceTemplate extends Schema.Class("MCP.ResourceTemplate")({ - server: ServerName, - name: Schema.String, - uriTemplate: Schema.String, - description: Schema.String.pipe(Schema.optional), - mimeType: Schema.String.pipe(Schema.optional), -}) {} - -export class ResourceCatalog extends Schema.Class("MCP.ResourceCatalog")({ - resources: Schema.Array(Resource), - templates: Schema.Array(ResourceTemplate), -}) {} - -export const ResourceContentPart = Schema.Union([ - Schema.Struct({ - type: Schema.Literal("text"), - uri: Schema.String, - text: Schema.String, - mimeType: Schema.String.pipe(Schema.optional), - }), - Schema.Struct({ - type: Schema.Literal("blob"), - uri: Schema.String, - blob: Schema.String, - mimeType: Schema.String.pipe(Schema.optional), - }), -]).pipe(Schema.toTaggedUnion("type")) -export type ResourceContentPart = typeof ResourceContentPart.Type - -export class ResourceContent extends Schema.Class("MCP.ResourceContent")({ - server: ServerName, - uri: Schema.String, - contents: Schema.Array(ResourceContentPart), -}) {} +export const Resource = Mcp.Resource +export type Resource = Mcp.Resource +export const ResourceTemplate = Mcp.ResourceTemplate +export type ResourceTemplate = Mcp.ResourceTemplate +export const ResourceCatalog = Mcp.ResourceCatalog +export type ResourceCatalog = Mcp.ResourceCatalog +export const ResourceContentPart = Mcp.ResourceContentPart +export type ResourceContentPart = Mcp.ResourceContentPart +export const ResourceContent = Mcp.ResourceContent +export type ResourceContent = Mcp.ResourceContent export class NotFoundError extends Schema.TaggedErrorClass()("MCP.NotFoundError", { server: ServerName, @@ -415,6 +383,24 @@ export const layer = Layer.effect( ), }) + const toResource = (server: ServerName, def: MCPClient.ResourceDefinition) => + Resource.make({ + server, + name: def.name, + uri: def.uri, + description: def.description, + mimeType: def.mimeType, + }) + + const toResourceTemplate = (server: ServerName, def: MCPClient.ResourceTemplateDefinition) => + ResourceTemplate.make({ + server, + name: def.name, + uriTemplate: def.uriTemplate, + description: def.description, + mimeType: def.mimeType, + }) + const refreshTools = (name: ServerName, entry: ServerEntry, connection: MCPClient.Connection) => connection.tools().pipe( Effect.map((defs) => { @@ -443,6 +429,7 @@ export const layer = Layer.effect( entry.prompts = undefined entry.status = { status: "failed", error: "Connection closed" } fork(events.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore)) + fork(events.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore)) fork(events.publish(Command.Event.Updated, {}).pipe(Effect.ignore)) fork(events.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)) }) @@ -458,6 +445,10 @@ export const layer = Layer.effect( connection.onPromptsChanged(() => { fork(refreshPrompts(name, entry, connection).pipe(Effect.ignore)) }) + connection.onResourcesChanged(() => { + if (entry.client !== connection) return + fork(events.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore)) + }) } const serverLog = (server: ServerName, message: MCPClient.LogMessage) => { @@ -501,6 +492,7 @@ export const layer = Layer.effect( // after the initial registration sweep and emits no list-changed notification would otherwise // stay invisible to the model. yield* events.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore) + yield* events.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore) yield* events.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) fork(refreshPrompts(name, entry, result.value.connection).pipe(Effect.ignore)) return @@ -557,11 +549,6 @@ export const layer = Layer.effect( concurrency: "unbounded", discard: true, }) - const gate = Effect.fnUntraced(function* (server: ServerName | string) { - const target = yield* requireServer(server) - yield* Deferred.await(target.entry.startup) - }) - return Service.of({ servers: Effect.fn("MCP.servers")(function* () { const entries = Array.from(runtime).toSorted(([a], [b]) => a.localeCompare(b)) @@ -637,11 +624,54 @@ export const layer = Layer.effect( }), resourceCatalog: Effect.fn("MCP.resourceCatalog")(function* () { yield* whenAllReady - return new ResourceCatalog({ resources: [], templates: [] }) + const catalogs = yield* Effect.forEach( + Array.from(runtime), + ([name, entry]) => { + if (!entry.client) return Effect.succeed({ resources: [], templates: [] }) + return Effect.all( + { + resources: entry.client.resources().pipe(Effect.catch(() => Effect.succeed([]))), + templates: entry.client.resourceTemplates().pipe(Effect.catch(() => Effect.succeed([]))), + }, + { concurrency: "unbounded" }, + ).pipe( + Effect.map((catalog) => ({ + resources: catalog.resources.map((def) => toResource(name, def)), + templates: catalog.templates.map((def) => toResourceTemplate(name, def)), + })), + ) + }, + { concurrency: "unbounded" }, + ) + return ResourceCatalog.make({ + resources: catalogs + .flatMap((catalog) => catalog.resources) + .toSorted( + (a, b) => a.server.localeCompare(b.server) || a.name.localeCompare(b.name) || a.uri.localeCompare(b.uri), + ), + templates: catalogs + .flatMap((catalog) => catalog.templates) + .toSorted( + (a, b) => + a.server.localeCompare(b.server) || + a.name.localeCompare(b.name) || + a.uriTemplate.localeCompare(b.uriTemplate), + ), + }) }), readResource: Effect.fn("MCP.readResource")(function* (input) { - yield* gate(input.server) - return undefined + const target = yield* requireServer(input.server) + yield* Deferred.await(target.entry.startup) + if (!target.entry.client) return undefined + const result = yield* target.entry.client + .readResource({ uri: input.uri }) + .pipe(Effect.catch(() => Effect.succeed(undefined))) + if (!result) return undefined + return ResourceContent.make({ + server: target.name, + uri: input.uri, + contents: result.contents, + }) }), }) }), diff --git a/packages/core/test/fixture/mcp-timeout.ts b/packages/core/test/fixture/mcp-timeout.ts index 15c851236c..4855b0ec26 100644 --- a/packages/core/test/fixture/mcp-timeout.ts +++ b/packages/core/test/fixture/mcp-timeout.ts @@ -4,16 +4,27 @@ import { CallToolRequestSchema, GetPromptRequestSchema, ListPromptsRequestSchema, + ListResourcesRequestSchema, + ListResourceTemplatesRequestSchema, ListToolsRequestSchema, + ReadResourceRequestSchema, } from "@modelcontextprotocol/sdk/types.js" -const server = new Server({ name: "timeout", version: "1.0.0" }, { capabilities: { prompts: {}, tools: {} } }) +const server = new Server( + { name: "timeout", version: "1.0.0" }, + { capabilities: { prompts: {}, resources: {}, tools: {} } }, +) server.setRequestHandler(ListToolsRequestSchema, async () => { if (process.env.MCP_TIMEOUT_TARGET === "catalog") await Bun.sleep(100) return { tools: [{ name: "slow", inputSchema: { type: "object" } }] } }) server.setRequestHandler(ListPromptsRequestSchema, () => Promise.resolve({ prompts: [{ name: "slow" }] })) +server.setRequestHandler(ListResourcesRequestSchema, async () => { + if (process.env.MCP_TIMEOUT_TARGET === "resource-catalog") await Bun.sleep(100) + return { resources: [{ name: "slow", uri: "test://slow" }] } +}) +server.setRequestHandler(ListResourceTemplatesRequestSchema, () => Promise.resolve({ resourceTemplates: [] })) server.setRequestHandler(CallToolRequestSchema, async () => { await Bun.sleep(100) return { content: [] } @@ -22,5 +33,9 @@ server.setRequestHandler(GetPromptRequestSchema, async () => { await Bun.sleep(100) return { messages: [] } }) +server.setRequestHandler(ReadResourceRequestSchema, async (request) => { + await Bun.sleep(100) + return { contents: [{ uri: request.params.uri, text: "slow" }] } +}) await server.connect(new StdioServerTransport()) diff --git a/packages/core/test/fixture/mcp.ts b/packages/core/test/fixture/mcp.ts index 2c4d1f86a6..b247a4d8b7 100644 --- a/packages/core/test/fixture/mcp.ts +++ b/packages/core/test/fixture/mcp.ts @@ -14,15 +14,12 @@ export const emptyMcpLayer = Layer.succeed( instructions: () => Effect.succeed([]), prompts: () => Effect.succeed([]), prompt: () => Effect.succeed(undefined), - resourceCatalog: () => Effect.succeed(new MCP.ResourceCatalog({ resources: [], templates: [] })), + resourceCatalog: () => Effect.succeed(MCP.ResourceCatalog.make({ resources: [], templates: [] })), readResource: () => Effect.succeed(undefined), }), ) -export const emptyConfigLayer = Layer.succeed( - Config.Service, - Config.Service.of({ entries: () => Effect.succeed([]) }), -) +export const emptyConfigLayer = Layer.succeed(Config.Service, Config.Service.of({ entries: () => Effect.succeed([]) })) export const testLocationLayer = Layer.succeed( Location.Service, diff --git a/packages/core/test/mcp.test.ts b/packages/core/test/mcp.test.ts index 26977a8729..32a0f94d88 100644 --- a/packages/core/test/mcp.test.ts +++ b/packages/core/test/mcp.test.ts @@ -3,26 +3,165 @@ import { describe, expect, test } from "bun:test" import { Client } from "@modelcontextprotocol/sdk/client/index.js" import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js" import { Server } from "@modelcontextprotocol/sdk/server/index.js" -import { CallToolRequestSchema, ListToolsRequestSchema } from "@modelcontextprotocol/sdk/types.js" +import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js" +import { + CallToolRequestSchema, + ListResourcesRequestSchema, + ListResourceTemplatesRequestSchema, + ListToolsRequestSchema, + ReadResourceRequestSchema, +} from "@modelcontextprotocol/sdk/types.js" import { ConfigMCP } from "@opencode-ai/core/config/mcp" +import { Config } from "@opencode-ai/core/config" +import { Credential } from "@opencode-ai/core/credential" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { EventV2 } from "@opencode-ai/core/event" +import { Form } from "@opencode-ai/core/form" +import { Integration } from "@opencode-ai/core/integration" +import { Location } from "@opencode-ai/core/location" import { MCP } from "@opencode-ai/core/mcp/index" import { MCPClient } from "@opencode-ai/core/mcp/client" import { PermissionV2 } from "@opencode-ai/core/permission" +import { AbsolutePath } from "@opencode-ai/core/schema" import { SessionV2 } from "@opencode-ai/core/session" import { McpTool } from "@opencode-ai/core/tool/mcp" import { ToolRegistry } from "@opencode-ai/core/tool/registry" import { ToolOutputStore } from "@opencode-ai/core/tool-output-store" -import { Deferred, Effect, Fiber, Layer, Stream } from "effect" +import { Deferred, Effect, Exit, Fiber, Layer, Stream } from "effect" import { testEffect } from "./lib/effect" +import { location } from "./fixture/location" import { settleTool, toolDefinitions, toolIdentity, waitForTool } from "./lib/tool" let assertion: Deferred.Deferred | undefined let decision: Effect.Effect = Effect.void let calls = 0 +type ResourcePage = { + items: Array<{ name: string; uri: string; description?: string; mimeType?: string }> + nextCursor?: string +} + +type ResourceTemplatePage = { + items: Array<{ name: string; uriTemplate: string; description?: string; mimeType?: string }> + nextCursor?: string +} + +function resourceServer(input: { resources?: boolean; listChanged?: boolean } = {}) { + return Effect.acquireRelease( + Effect.promise(async () => { + const state = { + resources: [] as ResourcePage["items"], + templates: [] as ResourceTemplatePage["items"], + resourcePages: undefined as Record | undefined, + templatePages: undefined as Record | undefined, + contents: [ + { uri: "docs://readme", text: "hello", mimeType: "text/plain" }, + { uri: "docs://logo", blob: "aGVsbG8=", mimeType: "image/png" }, + ] as Array<{ uri: string; text: string; mimeType?: string } | { uri: string; blob: string; mimeType?: string }>, + resourceLists: 0, + templateLists: 0, + } + const protocol = new Server( + { name: "mcp-resources", version: "1.0.0" }, + { + capabilities: { + tools: {}, + ...(input.resources === false ? {} : { resources: { listChanged: input.listChanged } }), + }, + }, + ) + protocol.setRequestHandler(ListToolsRequestSchema, () => Promise.resolve({ tools: [] })) + if (input.resources !== false) { + protocol.setRequestHandler(ListResourcesRequestSchema, (request) => { + state.resourceLists += 1 + const page = state.resourcePages?.[request.params?.cursor ?? "initial"] + return Promise.resolve({ resources: page?.items ?? state.resources, nextCursor: page?.nextCursor }) + }) + protocol.setRequestHandler(ListResourceTemplatesRequestSchema, (request) => { + state.templateLists += 1 + const page = state.templatePages?.[request.params?.cursor ?? "initial"] + return Promise.resolve({ resourceTemplates: page?.items ?? state.templates, nextCursor: page?.nextCursor }) + }) + protocol.setRequestHandler(ReadResourceRequestSchema, () => Promise.resolve({ contents: state.contents })) + } + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => crypto.randomUUID(), + enableJsonResponse: true, + }) + await protocol.connect(transport) + const http = Bun.serve({ + port: 0, + fetch: (request) => transport.handleRequest(request), + }) + return { + state, + url: http.url.toString(), + sendResourceListChanged: () => protocol.sendResourceListChanged(), + close: async () => { + await protocol.close().catch(() => {}) + await http.stop(true) + }, + } + }), + (server) => Effect.promise(server.close), + ) +} + +function resourceMcpLayer(url: string) { + const directory = AbsolutePath.make(import.meta.dir) + const unusedIntegration = () => Effect.die("unused integration service") + return MCP.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed( + Config.Service, + Config.Service.of({ + entries: () => + Effect.succeed([ + new Config.Document({ + type: "document", + info: new Config.Info({ + mcp: new ConfigMCP.Info({ + servers: { resources: new ConfigMCP.Remote({ type: "remote", url, oauth: false }) }, + }), + }), + }), + ]), + }), + ), + Layer.succeed(Location.Service, Location.Service.of(location({ directory }))), + Layer.mock(EventV2.Service, { + subscribe: () => Stream.never, + publish: (definition, data) => + Effect.succeed({ + id: EventV2.ID.create(), + type: definition.type, + data, + } as EventV2.Payload), + }), + Layer.mock(Form.Service, {}), + Layer.mock(Integration.Service, { + connection: { + active: unusedIntegration, + resolve: unusedIntegration, + key: unusedIntegration, + oauth: unusedIntegration, + update: unusedIntegration, + remove: unusedIntegration, + }, + attempt: { + status: unusedIntegration, + complete: unusedIntegration, + cancel: unusedIntegration, + }, + }), + Layer.mock(Credential.Service, {}), + ), + ), + ) +} + const mcp = Layer.mock(MCP.Service, { tools: () => Effect.succeed([ @@ -241,6 +380,163 @@ test("applies the configured MCP execution timeout to prompts", async () => { await expect(result).rejects.toThrow("Request timed out") }) +test("applies configured MCP timeouts to resource operations", async () => { + const catalog = Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const connection = yield* MCPClient.connect( + "resource-catalog-timeout", + new ConfigMCP.Local({ + type: "local", + command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-timeout.ts")], + environment: { MCP_TIMEOUT_TARGET: "resource-catalog" }, + timeout: new ConfigMCP.Timeout({ catalog: 10 }), + }), + import.meta.dir, + ) + return yield* connection.resources() + }), + ), + ) + await expect(catalog).rejects.toThrow("Request timed out") + + const read = Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const connection = yield* MCPClient.connect( + "resource-read-timeout", + new ConfigMCP.Local({ + type: "local", + command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-timeout.ts")], + timeout: new ConfigMCP.Timeout({ execution: 10 }), + }), + import.meta.dir, + ) + return yield* connection.readResource({ uri: "test://slow" }) + }), + ), + ) + await expect(read).rejects.toThrow("Request timed out") +}) + +test("lists, reads, and reports MCP resource changes", async () => { + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const server = yield* resourceServer({ listChanged: true }) + server.state.resourcePages = { + initial: { + items: [{ name: "Readme", uri: "docs://readme", description: "Project docs" }], + nextCursor: "resources-2", + }, + "resources-2": { items: [{ name: "Logo", uri: "docs://logo", mimeType: "image/png" }] }, + } + server.state.templatePages = { + initial: { + items: [{ name: "File", uriTemplate: "docs://{path}" }], + nextCursor: "templates-2", + }, + "templates-2": { items: [{ name: "Issue", uriTemplate: "issue://{id}", description: "Issue" }] }, + } + const connection = yield* MCPClient.connect( + "resources", + new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false }), + import.meta.dir, + ) + + expect(yield* connection.resources()).toEqual([ + { name: "Readme", uri: "docs://readme", description: "Project docs", mimeType: undefined }, + { name: "Logo", uri: "docs://logo", description: undefined, mimeType: "image/png" }, + ]) + expect(yield* connection.resourceTemplates()).toEqual([ + { name: "File", uriTemplate: "docs://{path}", description: undefined, mimeType: undefined }, + { name: "Issue", uriTemplate: "issue://{id}", description: "Issue", mimeType: undefined }, + ]) + expect(yield* connection.readResource({ uri: "docs://readme" })).toEqual({ + contents: [ + { type: "text", uri: "docs://readme", text: "hello", mimeType: "text/plain" }, + { type: "blob", uri: "docs://logo", blob: "aGVsbG8=", mimeType: "image/png" }, + ], + }) + + const changed = yield* Deferred.make() + connection.onResourcesChanged(() => Deferred.doneUnsafe(changed, Exit.void)) + yield* Effect.promise(server.sendResourceListChanged) + yield* Deferred.await(changed) + }), + ), + ) +}) + +test("skips MCP resource requests when the capability is absent", async () => { + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const server = yield* resourceServer({ resources: false }) + const connection = yield* MCPClient.connect( + "resources", + new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false }), + import.meta.dir, + ) + expect(yield* connection.resources()).toEqual([]) + expect(yield* connection.resourceTemplates()).toEqual([]) + expect(yield* connection.readResource({ uri: "docs://readme" })).toBeUndefined() + expect({ resources: server.state.resourceLists, templates: server.state.templateLists }).toEqual({ + resources: 0, + templates: 0, + }) + }), + ), + ) +}) + +test("loads and reads MCP resources", async () => { + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const server = yield* resourceServer() + server.state.resources = [{ name: "Readme", uri: "docs://readme" }] + server.state.templates = [{ name: "File", uriTemplate: "docs://{path}" }] + + yield* Effect.gen(function* () { + const service = yield* MCP.Service + expect(yield* service.resourceCatalog()).toEqual({ + resources: [ + { + server: "resources", + name: "Readme", + uri: "docs://readme", + description: undefined, + mimeType: undefined, + }, + ], + templates: [ + { + server: "resources", + name: "File", + uriTemplate: "docs://{path}", + description: undefined, + mimeType: undefined, + }, + ], + }) + + server.state.resources = [{ name: "Guide", uri: "docs://guide" }] + expect((yield* service.resourceCatalog()).resources.map((resource) => resource.uri)).toEqual(["docs://guide"]) + expect(yield* service.readResource({ server: "resources", uri: "docs://readme" })).toEqual({ + server: "resources", + uri: "docs://readme", + contents: [ + { type: "text", uri: "docs://readme", text: "hello", mimeType: "text/plain" }, + { type: "blob", uri: "docs://logo", blob: "aGVsbG8=", mimeType: "image/png" }, + ], + }) + }).pipe(Effect.provide(resourceMcpLayer(server.url))) + }), + ), + ) +}) + it.effect("advertises MCP output schemas to Code Mode", () => Effect.gen(function* () { const registry = yield* ToolRegistry.Service diff --git a/packages/schema/src/index.ts b/packages/schema/src/index.ts index 3f2efee68d..4e77cc4151 100644 --- a/packages/schema/src/index.ts +++ b/packages/schema/src/index.ts @@ -9,6 +9,7 @@ export { Form } from "./form.js" export { Integration } from "./integration.js" export { LLM } from "./llm.js" export { Location } from "./location.js" +export { Mcp } from "./mcp.js" export { Model } from "./model.js" export { Permission } from "./permission.js" export { PermissionSaved } from "./permission-saved.js" diff --git a/packages/schema/src/mcp-event.ts b/packages/schema/src/mcp-event.ts index ae1e82656d..d41f5f0c4a 100644 --- a/packages/schema/src/mcp-event.ts +++ b/packages/schema/src/mcp-event.ts @@ -10,6 +10,13 @@ export const ToolsChanged = Event.ephemeral({ }, }) +export const ResourcesChanged = Event.ephemeral({ + type: "mcp.resources.changed", + schema: { + server: Schema.String, + }, +}) + export const BrowserOpenFailed = Event.ephemeral({ type: "mcp.browser.open.failed", schema: { @@ -27,4 +34,4 @@ export const StatusChanged = Event.ephemeral({ }, }) -export const Definitions = Event.inventory(ToolsChanged, StatusChanged) +export const Definitions = Event.inventory(ToolsChanged, ResourcesChanged, StatusChanged) diff --git a/packages/schema/src/mcp.ts b/packages/schema/src/mcp.ts index 4a7ce15618..a86ffe6d67 100644 --- a/packages/schema/src/mcp.ts +++ b/packages/schema/src/mcp.ts @@ -25,14 +25,9 @@ const NeedsClientRegistration = Schema.Struct({ }).annotate({ identifier: "Mcp.Status.NeedsClientRegistration" }) export type Status = typeof Status.Type -export const Status = Schema.Union([ - Connected, - Pending, - Disabled, - Failed, - NeedsAuth, - NeedsClientRegistration, -]).pipe(Schema.toTaggedUnion("status")) +export const Status = Schema.Union([Connected, Pending, Disabled, Failed, NeedsAuth, NeedsClientRegistration]).pipe( + Schema.toTaggedUnion("status"), +) export interface Server extends Schema.Schema.Type {} export const Server = Schema.Struct({ @@ -42,3 +37,50 @@ export const Server = Schema.Struct({ // without matching by name, which could collide with provider or plugin integrations. integrationID: optional(IntegrationID), }).annotate({ identifier: "Mcp.Server" }) + +export interface Resource extends Schema.Schema.Type {} +export const Resource = Schema.Struct({ + server: Schema.String, + name: Schema.String, + uri: Schema.String, + description: optional(Schema.String), + mimeType: optional(Schema.String), +}).annotate({ identifier: "Mcp.Resource" }) + +export interface ResourceTemplate extends Schema.Schema.Type {} +export const ResourceTemplate = Schema.Struct({ + server: Schema.String, + name: Schema.String, + uriTemplate: Schema.String, + description: optional(Schema.String), + mimeType: optional(Schema.String), +}).annotate({ identifier: "Mcp.ResourceTemplate" }) + +export interface ResourceCatalog extends Schema.Schema.Type {} +export const ResourceCatalog = Schema.Struct({ + resources: Schema.Array(Resource), + templates: Schema.Array(ResourceTemplate), +}).annotate({ identifier: "Mcp.ResourceCatalog" }) + +export const ResourceContentPart = Schema.Union([ + Schema.Struct({ + type: Schema.Literal("text"), + uri: Schema.String, + text: Schema.String, + mimeType: optional(Schema.String), + }), + Schema.Struct({ + type: Schema.Literal("blob"), + uri: Schema.String, + blob: Schema.String, + mimeType: optional(Schema.String), + }), +]).pipe(Schema.toTaggedUnion("type"), Schema.annotate({ identifier: "Mcp.ResourceContentPart" })) +export type ResourceContentPart = typeof ResourceContentPart.Type + +export interface ResourceContent extends Schema.Schema.Type {} +export const ResourceContent = Schema.Struct({ + server: Schema.String, + uri: Schema.String, + contents: Schema.Array(ResourceContentPart), +}).annotate({ identifier: "Mcp.ResourceContent" }) diff --git a/packages/schema/test/contract-hygiene.test.ts b/packages/schema/test/contract-hygiene.test.ts index 236e15b91f..0cc1c71974 100644 --- a/packages/schema/test/contract-hygiene.test.ts +++ b/packages/schema/test/contract-hygiene.test.ts @@ -2,6 +2,7 @@ import { describe, expect, test } from "bun:test" import { DateTime, Schema } from "effect" import { Agent } from "../src/agent.js" import { FileSystem } from "../src/filesystem.js" +import { Mcp } from "../src/mcp.js" import { Model } from "../src/model.js" import { Project } from "../src/project.js" import { Provider } from "../src/provider.js" @@ -51,6 +52,11 @@ describe("contract hygiene", () => { const identifiers = [ Agent.Color, FileSystem.Submatch, + Mcp.Resource, + Mcp.ResourceTemplate, + Mcp.ResourceCatalog, + Mcp.ResourceContentPart, + Mcp.ResourceContent, Model.Ref, Model.Capabilities, Model.Cost, diff --git a/packages/schema/test/event-manifest.test.ts b/packages/schema/test/event-manifest.test.ts index 1a5cbe7fa7..f93fb89de7 100644 --- a/packages/schema/test/event-manifest.test.ts +++ b/packages/schema/test/event-manifest.test.ts @@ -53,6 +53,7 @@ describe("public event manifest", () => { expect(EventManifest.Server.get("mcp.status.changed")).toBe(McpEvent.StatusChanged) expect(EventManifest.Server.get("session.deleted")).toBe(SessionEvent.Deleted) expect(EventManifest.Server.has("mcp.tools.changed")).toBe(false) + expect(EventManifest.Server.has("mcp.resources.changed")).toBe(false) expect(Agent.Event.Updated.durable).toBeUndefined() expect(EventManifest.Durable.has("agent.updated")).toBe(false) }) @@ -76,7 +77,7 @@ describe("public event manifest", () => { expect(Form.Event.Definitions).toEqual([Form.Event.Created, Form.Event.Replied, Form.Event.Cancelled]) expect(Reference.Event.Definitions).toEqual([Reference.Event.Updated]) expect(Plugin.Event.Definitions).toEqual([Plugin.Event.Added, Plugin.Event.Updated]) - expect(McpEvent.Definitions).toEqual([McpEvent.ToolsChanged, McpEvent.StatusChanged]) + expect(McpEvent.Definitions).toEqual([McpEvent.ToolsChanged, McpEvent.ResourcesChanged, McpEvent.StatusChanged]) expect(EventManifest.Latest.has("mcp.browser.open.failed")).toBe(false) expect(EventManifest.Latest.has("ide.installed")).toBe(false) expect(IdeEvent.Definitions).toEqual([IdeEvent.Installed]) diff --git a/packages/schema/test/mcp.test.ts b/packages/schema/test/mcp.test.ts new file mode 100644 index 0000000000..75daa7fe1c --- /dev/null +++ b/packages/schema/test/mcp.test.ts @@ -0,0 +1,37 @@ +import { describe, expect, test } from "bun:test" +import { Schema } from "effect" +import { Mcp } from "../src/mcp.js" + +describe("Mcp resources", () => { + test("decodes resource catalogs and omits absent metadata", () => { + const value = Schema.decodeUnknownSync(Mcp.ResourceCatalog)({ + resources: [{ server: "docs", name: "Readme", uri: "docs://readme" }], + templates: [{ server: "docs", name: "File", uriTemplate: "docs://{path}" }], + }) + + expect(Schema.encodeSync(Mcp.ResourceCatalog)(value)).toEqual({ + resources: [{ server: "docs", name: "Readme", uri: "docs://readme" }], + templates: [{ server: "docs", name: "File", uriTemplate: "docs://{path}" }], + }) + }) + + test("preserves text and base64 blob contents", () => { + expect( + Schema.decodeUnknownSync(Mcp.ResourceContent)({ + server: "docs", + uri: "docs://readme", + contents: [ + { type: "text", uri: "docs://readme", text: "hello", mimeType: "text/plain" }, + { type: "blob", uri: "docs://logo", blob: "aGVsbG8=", mimeType: "image/png" }, + ], + }), + ).toEqual({ + server: "docs", + uri: "docs://readme", + contents: [ + { type: "text", uri: "docs://readme", text: "hello", mimeType: "text/plain" }, + { type: "blob", uri: "docs://logo", blob: "aGVsbG8=", mimeType: "image/png" }, + ], + }) + }) +})