mirror of
https://github.com/anomalyco/opencode.git
synced 2026-07-10 02:58:29 +00:00
feat(core): support MCP resources (#35656)
This commit is contained in:
parent
d0733f83bf
commit
66a8d830aa
11 changed files with 618 additions and 84 deletions
|
|
@ -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<PromptMessage>
|
||||
}
|
||||
|
||||
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<ResourceContentPart>
|
||||
}
|
||||
|
||||
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<ToolDefinition[], Error>
|
||||
/** Lists the server's prompts; returns [] when the server doesn't advertise prompt support, fails on a transport error. */
|
||||
readonly prompts: () => Effect.Effect<PromptDefinition[], Error>
|
||||
/** Lists the server's resources; returns [] when the server doesn't advertise resource support. */
|
||||
readonly resources: () => Effect.Effect<ResourceDefinition[], Error>
|
||||
/** Lists the server's resource templates; returns [] when the server doesn't advertise resource support. */
|
||||
readonly resourceTemplates: () => Effect.Effect<ResourceTemplateDefinition[], Error>
|
||||
/** Reads one resource; returns undefined when the server doesn't advertise resource support. */
|
||||
readonly readResource: (input: { readonly uri: string }) => Effect.Effect<ReadResourceResult | undefined, Error>
|
||||
/** 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) })
|
||||
|
|
|
|||
|
|
@ -83,48 +83,16 @@ export class PromptResult extends Schema.Class<PromptResult>("MCP.PromptResult")
|
|||
messages: Schema.Array(PromptMessage),
|
||||
}) {}
|
||||
|
||||
export class Resource extends Schema.Class<Resource>("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<ResourceTemplate>("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<ResourceCatalog>("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<ResourceContent>("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<NotFoundError>()("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,
|
||||
})
|
||||
}),
|
||||
})
|
||||
}),
|
||||
|
|
|
|||
|
|
@ -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())
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<PermissionV2.AssertInput> | undefined
|
||||
let decision: Effect.Effect<void, PermissionV2.Error> = 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<string, ResourcePage> | undefined,
|
||||
templatePages: undefined as Record<string, ResourceTemplatePage> | 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<typeof definition>),
|
||||
}),
|
||||
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<void>()
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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<typeof Server> {}
|
||||
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<typeof Resource> {}
|
||||
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<typeof ResourceTemplate> {}
|
||||
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<typeof ResourceCatalog> {}
|
||||
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<typeof ResourceContent> {}
|
||||
export const ResourceContent = Schema.Struct({
|
||||
server: Schema.String,
|
||||
uri: Schema.String,
|
||||
contents: Schema.Array(ResourceContentPart),
|
||||
}).annotate({ identifier: "Mcp.ResourceContent" })
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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])
|
||||
|
|
|
|||
37
packages/schema/test/mcp.test.ts
Normal file
37
packages/schema/test/mcp.test.ts
Normal file
|
|
@ -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" },
|
||||
],
|
||||
})
|
||||
})
|
||||
})
|
||||
Loading…
Add table
Add a link
Reference in a new issue