|
|
@@ -166,616 +166,615 @@ export const Options = Schema.Struct({
|
|
|
})
|
|
|
export type Options = typeof Options.Type
|
|
|
|
|
|
-export const layer = (options?: Options) => Layer.effect(
|
|
|
- Service,
|
|
|
- Effect.gen(function* () {
|
|
|
- const config = yield* Config.Service
|
|
|
- const location = yield* Location.Service
|
|
|
- const bus = yield* Bus.Service
|
|
|
- const forms = yield* Form.Service
|
|
|
- const integration = yield* Integration.Service
|
|
|
- const credentials = yield* Credential.Service
|
|
|
- const root = yield* Scope.make()
|
|
|
- const fork = yield* FiberSet.makeRuntime<never, void, never>()
|
|
|
- yield* Effect.addFinalizer((exit) => Scope.close(root, exit))
|
|
|
-
|
|
|
- const documents = (yield* config.entries()).filter((entry): entry is Document => entry.type === "document")
|
|
|
- // Global MCP timeout defaults, later config files overriding earlier ones.
|
|
|
- const timeout = Object.assign(
|
|
|
- {},
|
|
|
- ...documents.flatMap((entry) => (entry.info.mcp?.timeout ? [entry.info.mcp.timeout] : [])),
|
|
|
- )
|
|
|
- // Later config files win for duplicate server names; per-server timeout overrides globals.
|
|
|
- const runtime = new Map<ServerName, ServerEntry>()
|
|
|
- // Serializes lifecycle operations per server. Anything taking this lock from a connection
|
|
|
- // callback must stay forked: lifecycle operations close scopes while holding it, firing onClose.
|
|
|
- const locks = KeyedMutex.makeUnsafe<ServerName>()
|
|
|
- const urlElicitations = new Map<string, Form.ID>()
|
|
|
- for (const entry of documents) {
|
|
|
- for (const [name, server] of Object.entries(entry.info.mcp?.servers ?? {})) {
|
|
|
- runtime.set(ServerName.make(name), {
|
|
|
- config: { ...server, timeout: { ...timeout, ...server.timeout } },
|
|
|
- status: { status: "pending" },
|
|
|
- startup: Deferred.makeUnsafe<void>(),
|
|
|
- })
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- // Register every remote server as an OAuth integration so credentials live in the global store
|
|
|
- // rather than in committed config. Servers that connect anonymously simply never use the method.
|
|
|
- const owned = new Set<Integration.ID>()
|
|
|
- const register = Effect.fnUntraced(function* (name: ServerName, entry: ServerEntry) {
|
|
|
- if (entry.config.type !== "remote" || entry.config.oauth === false) return
|
|
|
- const remote = entry.config
|
|
|
- // Key identity on name + url, not url alone: two configs for the same url under different names are
|
|
|
- // distinct logical servers that may hold different accounts, so they must not share a credential row.
|
|
|
- const suffix =
|
|
|
- "mcp_" +
|
|
|
- createHash("sha1")
|
|
|
- .update(name + "\u0000" + remote.url)
|
|
|
- .digest("hex")
|
|
|
- .slice(0, 16)
|
|
|
- const integrationID = Integration.ID.make(suffix)
|
|
|
- entry.integrationID = integrationID
|
|
|
- owned.add(integrationID)
|
|
|
- const methodID = Integration.MethodID.make(suffix)
|
|
|
- // Each registration gets its own child scope so disposal detaches it from the root scope
|
|
|
- // entirely; registering directly on root would accumulate a dead finalizer per replaced or
|
|
|
- // removed server for the lifetime of the layer.
|
|
|
- const scope = yield* Scope.fork(root)
|
|
|
- entry.registration = { dispose: Scope.close(scope, Exit.void) }
|
|
|
- yield* integration
|
|
|
- .transform((draft) => {
|
|
|
- draft.update(integrationID, (ref) => {
|
|
|
- ref.name = name
|
|
|
- })
|
|
|
- draft.method.update({
|
|
|
- integrationID,
|
|
|
- method: { id: methodID, type: "oauth", label: name },
|
|
|
- authorize: () => MCPOAuth.authorize({ name, config: remote, methodID }),
|
|
|
+export const layer = (options?: Options) =>
|
|
|
+ Layer.effect(
|
|
|
+ Service,
|
|
|
+ Effect.gen(function* () {
|
|
|
+ const config = yield* Config.Service
|
|
|
+ const location = yield* Location.Service
|
|
|
+ const bus = yield* Bus.Service
|
|
|
+ const forms = yield* Form.Service
|
|
|
+ const integration = yield* Integration.Service
|
|
|
+ const credentials = yield* Credential.Service
|
|
|
+ const root = yield* Scope.make()
|
|
|
+ const fork = yield* FiberSet.makeRuntime<never, void, never>()
|
|
|
+ yield* Effect.addFinalizer((exit) => Scope.close(root, exit))
|
|
|
+
|
|
|
+ const documents = (yield* config.entries()).filter((entry): entry is Document => entry.type === "document")
|
|
|
+ // Global MCP timeout defaults, later config files overriding earlier ones.
|
|
|
+ const timeout = Object.assign(
|
|
|
+ {},
|
|
|
+ ...documents.flatMap((entry) => (entry.info.mcp?.timeout ? [entry.info.mcp.timeout] : [])),
|
|
|
+ )
|
|
|
+ // Later config files win for duplicate server names; per-server timeout overrides globals.
|
|
|
+ const runtime = new Map<ServerName, ServerEntry>()
|
|
|
+ // Serializes lifecycle operations per server. Anything taking this lock from a connection
|
|
|
+ // callback must stay forked: lifecycle operations close scopes while holding it, firing onClose.
|
|
|
+ const locks = KeyedMutex.makeUnsafe<ServerName>()
|
|
|
+ const urlElicitations = new Map<string, Form.ID>()
|
|
|
+ for (const entry of documents) {
|
|
|
+ for (const [name, server] of Object.entries(entry.info.mcp?.servers ?? {})) {
|
|
|
+ runtime.set(ServerName.make(name), {
|
|
|
+ config: { ...server, timeout: { ...timeout, ...server.timeout } },
|
|
|
+ status: { status: "pending" },
|
|
|
+ startup: Deferred.makeUnsafe<void>(),
|
|
|
})
|
|
|
- })
|
|
|
- .pipe(Scope.provide(scope))
|
|
|
- })
|
|
|
- yield* Effect.forEach(runtime, ([name, entry]) => register(name, entry), { discard: true })
|
|
|
-
|
|
|
- const requireServer = Effect.fnUntraced(function* (server: ServerName | string) {
|
|
|
- const name = ServerName.make(server)
|
|
|
- const entry = runtime.get(name)
|
|
|
- if (!entry) return yield* new NotFoundError({ server: name })
|
|
|
- return { name, entry }
|
|
|
- })
|
|
|
-
|
|
|
- // Builds the connect-time auth provider for a remote OAuth-integration server. The SDK presents and
|
|
|
- // refreshes stored tokens, persisting refreshes back to the same credential row. The provider never
|
|
|
- // opens a browser, so an auth-gated connect ends in UnauthorizedError -> needs_auth rather than a redirect.
|
|
|
- const connectProvider = Effect.fnUntraced(function* (entry: ServerEntry) {
|
|
|
- if (entry.config.type !== "remote" || !entry.integrationID) return undefined
|
|
|
- const remote = entry.config
|
|
|
- const oauth = remote.oauth || undefined
|
|
|
- const base = {
|
|
|
- redirectUrl: oauth?.redirect_uri ?? "http://127.0.0.1/callback",
|
|
|
- scope: oauth?.scope,
|
|
|
- client: oauth?.client_id ? { id: oauth.client_id, secret: oauth.client_secret } : undefined,
|
|
|
- // No browser during connect: an auth-gated server surfaces needs_auth instead of opening a browser.
|
|
|
- onRedirect: () => {},
|
|
|
+ }
|
|
|
}
|
|
|
- const stored = yield* credentials.list(entry.integrationID)
|
|
|
- const found = stored.find((credential) => credential.value.type === "oauth")
|
|
|
- if (!found || found.value.type !== "oauth")
|
|
|
- // No stored credential yet: an empty in-memory store still lets the SDK run the auth handshake, which
|
|
|
- // ends in UnauthorizedError -> needs_auth. Returning no provider instead would let the transport throw
|
|
|
- // a raw HTTP error, hiding the auth requirement behind a generic failed status. Anonymous servers are
|
|
|
- // unaffected: tokens() returns undefined, so no auth header is sent and the SDK never calls auth().
|
|
|
- return MCPOAuth.provider({ ...base, store: MCPOAuth.memoryStore() })
|
|
|
- const credentialID = found.id
|
|
|
- const methodID = found.value.methodID
|
|
|
- let current: Credential.OAuth | undefined = found.value
|
|
|
- return MCPOAuth.provider({
|
|
|
- ...base,
|
|
|
- // Drop a credential the SDK rejected so the next connect cleanly reports needs_auth. Uses the raw
|
|
|
- // credential service (no integration event) to avoid re-triggering the reconnect subscriber mid-connect.
|
|
|
- invalidate: async (scope) => {
|
|
|
- if (scope === "verifier" || scope === "discovery") return
|
|
|
- current = undefined
|
|
|
- await Effect.runPromise(credentials.remove(credentialID))
|
|
|
- },
|
|
|
- store: {
|
|
|
- tokens: async () => (current ? MCPOAuth.toTokens(current) : undefined),
|
|
|
- saveTokens: async (tokens) => {
|
|
|
- current = MCPOAuth.toCredential({
|
|
|
- methodID,
|
|
|
- serverUrl: remote.url,
|
|
|
- tokens,
|
|
|
- client: current ? MCPOAuth.clientFromCredential(current) : undefined,
|
|
|
+
|
|
|
+ // Register every remote server as an OAuth integration so credentials live in the global store
|
|
|
+ // rather than in committed config. Servers that connect anonymously simply never use the method.
|
|
|
+ const owned = new Set<Integration.ID>()
|
|
|
+ const register = Effect.fnUntraced(function* (name: ServerName, entry: ServerEntry) {
|
|
|
+ if (entry.config.type !== "remote" || entry.config.oauth === false) return
|
|
|
+ const remote = entry.config
|
|
|
+ // Key identity on name + url, not url alone: two configs for the same url under different names are
|
|
|
+ // distinct logical servers that may hold different accounts, so they must not share a credential row.
|
|
|
+ const suffix =
|
|
|
+ "mcp_" +
|
|
|
+ createHash("sha1")
|
|
|
+ .update(name + "\u0000" + remote.url)
|
|
|
+ .digest("hex")
|
|
|
+ .slice(0, 16)
|
|
|
+ const integrationID = Integration.ID.make(suffix)
|
|
|
+ entry.integrationID = integrationID
|
|
|
+ owned.add(integrationID)
|
|
|
+ const methodID = Integration.MethodID.make(suffix)
|
|
|
+ // Each registration gets its own child scope so disposal detaches it from the root scope
|
|
|
+ // entirely; registering directly on root would accumulate a dead finalizer per replaced or
|
|
|
+ // removed server for the lifetime of the layer.
|
|
|
+ const scope = yield* Scope.fork(root)
|
|
|
+ entry.registration = { dispose: Scope.close(scope, Exit.void) }
|
|
|
+ yield* integration
|
|
|
+ .transform((draft) => {
|
|
|
+ draft.update(integrationID, (ref) => {
|
|
|
+ ref.name = name
|
|
|
+ })
|
|
|
+ draft.method.update({
|
|
|
+ integrationID,
|
|
|
+ method: { id: methodID, type: "oauth", label: name },
|
|
|
+ authorize: () => MCPOAuth.authorize({ name, config: remote, methodID }),
|
|
|
})
|
|
|
- await Effect.runPromise(credentials.update(credentialID, { value: current }))
|
|
|
+ })
|
|
|
+ .pipe(Scope.provide(scope))
|
|
|
+ })
|
|
|
+ yield* Effect.forEach(runtime, ([name, entry]) => register(name, entry), { discard: true })
|
|
|
+
|
|
|
+ const requireServer = Effect.fnUntraced(function* (server: ServerName | string) {
|
|
|
+ const name = ServerName.make(server)
|
|
|
+ const entry = runtime.get(name)
|
|
|
+ if (!entry) return yield* new NotFoundError({ server: name })
|
|
|
+ return { name, entry }
|
|
|
+ })
|
|
|
+
|
|
|
+ // Builds the connect-time auth provider for a remote OAuth-integration server. The SDK presents and
|
|
|
+ // refreshes stored tokens, persisting refreshes back to the same credential row. The provider never
|
|
|
+ // opens a browser, so an auth-gated connect ends in UnauthorizedError -> needs_auth rather than a redirect.
|
|
|
+ const connectProvider = Effect.fnUntraced(function* (entry: ServerEntry) {
|
|
|
+ if (entry.config.type !== "remote" || !entry.integrationID) return undefined
|
|
|
+ const remote = entry.config
|
|
|
+ const oauth = remote.oauth || undefined
|
|
|
+ const base = {
|
|
|
+ redirectUrl: oauth?.redirect_uri ?? "http://127.0.0.1/callback",
|
|
|
+ scope: oauth?.scope,
|
|
|
+ client: oauth?.client_id ? { id: oauth.client_id, secret: oauth.client_secret } : undefined,
|
|
|
+ // No browser during connect: an auth-gated server surfaces needs_auth instead of opening a browser.
|
|
|
+ onRedirect: () => {},
|
|
|
+ }
|
|
|
+ const stored = yield* credentials.list(entry.integrationID)
|
|
|
+ const found = stored.find((credential) => credential.value.type === "oauth")
|
|
|
+ if (!found || found.value.type !== "oauth")
|
|
|
+ // No stored credential yet: an empty in-memory store still lets the SDK run the auth handshake, which
|
|
|
+ // ends in UnauthorizedError -> needs_auth. Returning no provider instead would let the transport throw
|
|
|
+ // a raw HTTP error, hiding the auth requirement behind a generic failed status. Anonymous servers are
|
|
|
+ // unaffected: tokens() returns undefined, so no auth header is sent and the SDK never calls auth().
|
|
|
+ return MCPOAuth.provider({ ...base, store: MCPOAuth.memoryStore() })
|
|
|
+ const credentialID = found.id
|
|
|
+ const methodID = found.value.methodID
|
|
|
+ let current: Credential.OAuth | undefined = found.value
|
|
|
+ return MCPOAuth.provider({
|
|
|
+ ...base,
|
|
|
+ // Drop a credential the SDK rejected so the next connect cleanly reports needs_auth. Uses the raw
|
|
|
+ // credential service (no integration event) to avoid re-triggering the reconnect subscriber mid-connect.
|
|
|
+ invalidate: async (scope) => {
|
|
|
+ if (scope === "verifier" || scope === "discovery") return
|
|
|
+ current = undefined
|
|
|
+ await Effect.runPromise(credentials.remove(credentialID))
|
|
|
},
|
|
|
- clientInformation: async () => (current ? MCPOAuth.clientFromCredential(current) : undefined),
|
|
|
- saveClientInformation: async () => {},
|
|
|
- codeVerifier: async () => undefined,
|
|
|
- saveCodeVerifier: async () => {},
|
|
|
- },
|
|
|
+ store: {
|
|
|
+ tokens: async () => (current ? MCPOAuth.toTokens(current) : undefined),
|
|
|
+ saveTokens: async (tokens) => {
|
|
|
+ current = MCPOAuth.toCredential({
|
|
|
+ methodID,
|
|
|
+ serverUrl: remote.url,
|
|
|
+ tokens,
|
|
|
+ client: current ? MCPOAuth.clientFromCredential(current) : undefined,
|
|
|
+ })
|
|
|
+ await Effect.runPromise(credentials.update(credentialID, { value: current }))
|
|
|
+ },
|
|
|
+ clientInformation: async () => (current ? MCPOAuth.clientFromCredential(current) : undefined),
|
|
|
+ saveClientInformation: async () => {},
|
|
|
+ codeVerifier: async () => undefined,
|
|
|
+ saveCodeVerifier: async () => {},
|
|
|
+ },
|
|
|
+ })
|
|
|
})
|
|
|
- })
|
|
|
-
|
|
|
- const elicitation = {
|
|
|
- create: (input: {
|
|
|
- readonly server: string
|
|
|
- readonly params: MCPClient.ElicitationParams
|
|
|
- readonly signal: AbortSignal
|
|
|
- }) =>
|
|
|
- Effect.gen(function* () {
|
|
|
- if (input.params.mode === "url") {
|
|
|
- const formID = Form.ID.create()
|
|
|
- const key = input.server + "\u0000" + input.params.elicitationId
|
|
|
- urlElicitations.set(key, formID)
|
|
|
+
|
|
|
+ const elicitation = {
|
|
|
+ create: (input: {
|
|
|
+ readonly server: string
|
|
|
+ readonly params: MCPClient.ElicitationParams
|
|
|
+ readonly signal: AbortSignal
|
|
|
+ }) =>
|
|
|
+ Effect.gen(function* () {
|
|
|
+ if (input.params.mode === "url") {
|
|
|
+ const formID = Form.ID.create()
|
|
|
+ const key = input.server + "\u0000" + input.params.elicitationId
|
|
|
+ urlElicitations.set(key, formID)
|
|
|
+ return yield* forms
|
|
|
+ .ask({
|
|
|
+ id: formID,
|
|
|
+ sessionID: GLOBAL_ELICITATION_SESSION_ID,
|
|
|
+ title: `${input.server} is requesting input`,
|
|
|
+ metadata: {
|
|
|
+ kind: "mcp-elicitation",
|
|
|
+ server: input.server,
|
|
|
+ elicitationID: input.params.elicitationId,
|
|
|
+ message: input.params.message,
|
|
|
+ },
|
|
|
+ fields: [{ key: URL_ELICITATION_FIELD_KEY, type: "external", url: input.params.url }],
|
|
|
+ })
|
|
|
+ .pipe(
|
|
|
+ Effect.raceFirst(waitForAbort(input.signal)),
|
|
|
+ Effect.ensuring(Effect.sync(() => urlElicitations.delete(key))),
|
|
|
+ Effect.map(
|
|
|
+ (state): MCPClient.ElicitationResult => ({
|
|
|
+ action: state.status === "answered" ? "accept" : "cancel",
|
|
|
+ }),
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ }
|
|
|
+ const params = input.params
|
|
|
+ const [field, ...fields] = Object.entries(params.requestedSchema.properties).map(([key, property]) =>
|
|
|
+ toElicitationField(key, property, params.requestedSchema.required?.includes(key) === true),
|
|
|
+ )
|
|
|
+ if (!field) return { action: "accept", content: {} }
|
|
|
return yield* forms
|
|
|
.ask({
|
|
|
- id: formID,
|
|
|
sessionID: GLOBAL_ELICITATION_SESSION_ID,
|
|
|
title: `${input.server} is requesting input`,
|
|
|
- metadata: {
|
|
|
- kind: "mcp-elicitation",
|
|
|
- server: input.server,
|
|
|
- elicitationID: input.params.elicitationId,
|
|
|
- message: input.params.message,
|
|
|
- },
|
|
|
- fields: [{ key: URL_ELICITATION_FIELD_KEY, type: "external", url: input.params.url }],
|
|
|
+ metadata: { kind: "mcp-elicitation", server: input.server, message: params.message },
|
|
|
+ fields: [field, ...fields],
|
|
|
})
|
|
|
.pipe(
|
|
|
Effect.raceFirst(waitForAbort(input.signal)),
|
|
|
- Effect.ensuring(Effect.sync(() => urlElicitations.delete(key))),
|
|
|
- Effect.map(
|
|
|
- (state): MCPClient.ElicitationResult => ({
|
|
|
- action: state.status === "answered" ? "accept" : "cancel",
|
|
|
- }),
|
|
|
- ),
|
|
|
- )
|
|
|
- }
|
|
|
- const params = input.params
|
|
|
- const [field, ...fields] = Object.entries(params.requestedSchema.properties).map(([key, property]) =>
|
|
|
- toElicitationField(key, property, params.requestedSchema.required?.includes(key) === true),
|
|
|
- )
|
|
|
- if (!field) return { action: "accept", content: {} }
|
|
|
- return yield* forms
|
|
|
- .ask({
|
|
|
- sessionID: GLOBAL_ELICITATION_SESSION_ID,
|
|
|
- title: `${input.server} is requesting input`,
|
|
|
- metadata: { kind: "mcp-elicitation", server: input.server, message: params.message },
|
|
|
- fields: [field, ...fields],
|
|
|
- })
|
|
|
- .pipe(
|
|
|
- Effect.raceFirst(waitForAbort(input.signal)),
|
|
|
- Effect.map((state): MCPClient.ElicitationResult => {
|
|
|
- if (state.status !== "answered") return { action: "cancel" }
|
|
|
- return {
|
|
|
- action: "accept",
|
|
|
- content: Object.fromEntries(
|
|
|
- Object.entries(state.answer).map(
|
|
|
- ([key, value]): [string, NonNullable<MCPClient.ElicitationResult["content"]>[string]] =>
|
|
|
- typeof value === "object" ? [key, Array.from(value)] : [key, value],
|
|
|
+ Effect.map((state): MCPClient.ElicitationResult => {
|
|
|
+ if (state.status !== "answered") return { action: "cancel" }
|
|
|
+ return {
|
|
|
+ action: "accept",
|
|
|
+ content: Object.fromEntries(
|
|
|
+ Object.entries(state.answer).map(
|
|
|
+ ([key, value]): [string, NonNullable<MCPClient.ElicitationResult["content"]>[string]] =>
|
|
|
+ typeof value === "object" ? [key, Array.from(value)] : [key, value],
|
|
|
+ ),
|
|
|
),
|
|
|
- ),
|
|
|
- }
|
|
|
+ }
|
|
|
+ }),
|
|
|
+ )
|
|
|
+ }),
|
|
|
+ complete: (input: { readonly server: string; readonly elicitationID: string }) =>
|
|
|
+ Effect.gen(function* () {
|
|
|
+ const formID = urlElicitations.get(input.server + "\u0000" + input.elicitationID)
|
|
|
+ if (!formID) return
|
|
|
+ yield* forms.reply({ id: formID, answer: { [URL_ELICITATION_FIELD_KEY]: true } }).pipe(Effect.ignore)
|
|
|
+ }),
|
|
|
+ } satisfies MCPClient.ElicitationHandler
|
|
|
+
|
|
|
+ const toTool = (server: ServerName, entry: ServerEntry, def: MCPClient.ToolDefinition) =>
|
|
|
+ new Tool({
|
|
|
+ server,
|
|
|
+ name: def.name,
|
|
|
+ codemode: entry.config.codemode,
|
|
|
+ description: def.description,
|
|
|
+ inputSchema: def.inputSchema,
|
|
|
+ outputSchema: def.outputSchema,
|
|
|
+ })
|
|
|
+
|
|
|
+ const toPrompt = (server: ServerName, def: MCPClient.PromptDefinition) =>
|
|
|
+ new Prompt({
|
|
|
+ server,
|
|
|
+ name: def.name,
|
|
|
+ description: def.description,
|
|
|
+ arguments: def.arguments?.map(
|
|
|
+ (argument) =>
|
|
|
+ new PromptArgument({
|
|
|
+ name: argument.name,
|
|
|
+ description: argument.description,
|
|
|
+ required: argument.required,
|
|
|
}),
|
|
|
- )
|
|
|
- }),
|
|
|
- complete: (input: { readonly server: string; readonly elicitationID: string }) =>
|
|
|
- Effect.gen(function* () {
|
|
|
- const formID = urlElicitations.get(input.server + "\u0000" + input.elicitationID)
|
|
|
- if (!formID) return
|
|
|
- yield* forms.reply({ id: formID, answer: { [URL_ELICITATION_FIELD_KEY]: true } }).pipe(Effect.ignore)
|
|
|
- }),
|
|
|
- } satisfies MCPClient.ElicitationHandler
|
|
|
-
|
|
|
- const toTool = (server: ServerName, entry: ServerEntry, def: MCPClient.ToolDefinition) =>
|
|
|
- new Tool({
|
|
|
- server,
|
|
|
- name: def.name,
|
|
|
- codemode: entry.config.codemode,
|
|
|
- description: def.description,
|
|
|
- inputSchema: def.inputSchema,
|
|
|
- outputSchema: def.outputSchema,
|
|
|
- })
|
|
|
+ ),
|
|
|
+ })
|
|
|
|
|
|
- const toPrompt = (server: ServerName, def: MCPClient.PromptDefinition) =>
|
|
|
- new Prompt({
|
|
|
- server,
|
|
|
- name: def.name,
|
|
|
- description: def.description,
|
|
|
- arguments: def.arguments?.map(
|
|
|
- (argument) =>
|
|
|
- new PromptArgument({
|
|
|
- name: argument.name,
|
|
|
- description: argument.description,
|
|
|
- required: argument.required,
|
|
|
- }),
|
|
|
- ),
|
|
|
- })
|
|
|
+ const toResource = (server: ServerName, def: MCPClient.ResourceDefinition) =>
|
|
|
+ Resource.make({
|
|
|
+ server,
|
|
|
+ name: def.name,
|
|
|
+ uri: def.uri,
|
|
|
+ description: def.description,
|
|
|
+ mimeType: def.mimeType,
|
|
|
+ })
|
|
|
|
|
|
- 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 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) => {
|
|
|
+ entry.tools = defs.map((def) => toTool(name, entry, def))
|
|
|
+ }),
|
|
|
+ )
|
|
|
|
|
|
- const refreshTools = (name: ServerName, entry: ServerEntry, connection: MCPClient.Connection) =>
|
|
|
- connection.tools().pipe(
|
|
|
- Effect.map((defs) => {
|
|
|
- entry.tools = defs.map((def) => toTool(name, entry, def))
|
|
|
- }),
|
|
|
- )
|
|
|
+ const refreshPrompts = (name: ServerName, entry: ServerEntry, connection: MCPClient.Connection) =>
|
|
|
+ connection.prompts().pipe(
|
|
|
+ Effect.map((defs) => {
|
|
|
+ entry.prompts = defs.map((def) => toPrompt(name, def))
|
|
|
+ }),
|
|
|
+ Effect.andThen(bus.publish(Command.Event.Updated, {})),
|
|
|
+ Effect.catch(() =>
|
|
|
+ Effect.sync(() => (entry.prompts = [])).pipe(Effect.andThen(bus.publish(Command.Event.Updated, {}))),
|
|
|
+ ),
|
|
|
+ )
|
|
|
|
|
|
- const refreshPrompts = (name: ServerName, entry: ServerEntry, connection: MCPClient.Connection) =>
|
|
|
- connection.prompts().pipe(
|
|
|
- Effect.map((defs) => {
|
|
|
- entry.prompts = defs.map((def) => toPrompt(name, def))
|
|
|
- }),
|
|
|
- Effect.andThen(bus.publish(Command.Event.Updated, {})),
|
|
|
- Effect.catch(() =>
|
|
|
- Effect.sync(() => (entry.prompts = [])).pipe(Effect.andThen(bus.publish(Command.Event.Updated, {}))),
|
|
|
- ),
|
|
|
- )
|
|
|
+ // Runs a connection callback under the server lock, dropping it if the connection is no longer
|
|
|
+ // the entry's live client, so late SDK callbacks cannot commit obsolete state.
|
|
|
+ const whenLive =
|
|
|
+ (name: ServerName, entry: ServerEntry, connection: MCPClient.Connection) =>
|
|
|
+ <E>(effect: Effect.Effect<void, E>) =>
|
|
|
+ fork(
|
|
|
+ Effect.suspend(() => (entry.client === connection ? effect : Effect.void)).pipe(
|
|
|
+ locks.withLock(name),
|
|
|
+ Effect.ignore,
|
|
|
+ ),
|
|
|
+ )
|
|
|
|
|
|
- // Runs a connection callback under the server lock, dropping it if the connection is no longer
|
|
|
- // the entry's live client, so late SDK callbacks cannot commit obsolete state.
|
|
|
- const whenLive =
|
|
|
- (name: ServerName, entry: ServerEntry, connection: MCPClient.Connection) =>
|
|
|
- <E>(effect: Effect.Effect<void, E>) =>
|
|
|
- fork(
|
|
|
- Effect.suspend(() => (entry.client === connection ? effect : Effect.void)).pipe(
|
|
|
- locks.withLock(name),
|
|
|
- Effect.ignore,
|
|
|
+ const watch = (name: ServerName, entry: ServerEntry, connection: MCPClient.Connection) => {
|
|
|
+ const live = whenLive(name, entry, connection)
|
|
|
+ connection.onClose(() =>
|
|
|
+ live(
|
|
|
+ Effect.gen(function* () {
|
|
|
+ entry.client = undefined
|
|
|
+ entry.tools = undefined
|
|
|
+ entry.prompts = undefined
|
|
|
+ entry.status = { status: "failed", error: "Connection closed" }
|
|
|
+ yield* bus.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore)
|
|
|
+ yield* bus.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore)
|
|
|
+ yield* bus.publish(Command.Event.Updated, {}).pipe(Effect.ignore)
|
|
|
+ yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
|
|
|
+ }),
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ connection.onLog((message) => fork(serverLog(name, message).pipe(Effect.ignore)))
|
|
|
+ connection.onToolsChanged(() =>
|
|
|
+ live(
|
|
|
+ refreshTools(name, entry, connection).pipe(
|
|
|
+ Effect.andThen(bus.publish(McpEvent.ToolsChanged, { server: name })),
|
|
|
+ ),
|
|
|
),
|
|
|
)
|
|
|
+ connection.onPromptsChanged(() => live(refreshPrompts(name, entry, connection)))
|
|
|
+ connection.onResourcesChanged(() => live(bus.publish(McpEvent.ResourcesChanged, { server: name })))
|
|
|
+ }
|
|
|
|
|
|
- const watch = (name: ServerName, entry: ServerEntry, connection: MCPClient.Connection) => {
|
|
|
- const live = whenLive(name, entry, connection)
|
|
|
- connection.onClose(() =>
|
|
|
- live(
|
|
|
- Effect.gen(function* () {
|
|
|
- entry.client = undefined
|
|
|
- entry.tools = undefined
|
|
|
- entry.prompts = undefined
|
|
|
- entry.status = { status: "failed", error: "Connection closed" }
|
|
|
+ const serverLog = (server: ServerName, message: MCPClient.LogMessage) => {
|
|
|
+ const fields = { server, logger: message.logger, level: message.level, data: message.data }
|
|
|
+ switch (message.level) {
|
|
|
+ case "debug":
|
|
|
+ return Effect.logDebug("MCP server log", fields)
|
|
|
+ case "info":
|
|
|
+ case "notice":
|
|
|
+ return Effect.logInfo("MCP server log", fields)
|
|
|
+ case "warning":
|
|
|
+ return Effect.logWarning("MCP server log", fields)
|
|
|
+ case "error":
|
|
|
+ case "critical":
|
|
|
+ case "alert":
|
|
|
+ case "emergency":
|
|
|
+ return Effect.logError("MCP server log", fields)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ const startServer = (name: ServerName, entry: ServerEntry) =>
|
|
|
+ Effect.gen(function* () {
|
|
|
+ // Announce the handshake so connect() and credential reconnects don't show a stale
|
|
|
+ // disabled/failed status for the duration of the connection attempt.
|
|
|
+ entry.status = { status: "pending" }
|
|
|
+ yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
|
|
|
+ const scope = yield* Scope.fork(root)
|
|
|
+ entry.scope = scope
|
|
|
+ const authProvider = yield* connectProvider(entry)
|
|
|
+ // List tools as part of connect so a failure here marks the server failed rather than
|
|
|
+ // leaving it connected with a silently empty tool list and no path to recover.
|
|
|
+ const result = yield* MCPClient.connect(
|
|
|
+ name,
|
|
|
+ entry.config,
|
|
|
+ location.directory,
|
|
|
+ authProvider,
|
|
|
+ elicitation,
|
|
|
+ options?.clientInfo,
|
|
|
+ ).pipe(
|
|
|
+ Effect.flatMap((connection) => connection.tools().pipe(Effect.map((tools) => ({ connection, tools })))),
|
|
|
+ Scope.provide(scope),
|
|
|
+ Effect.exit,
|
|
|
+ )
|
|
|
+ if (Exit.isSuccess(result)) {
|
|
|
+ entry.client = result.value.connection
|
|
|
+ entry.tools = result.value.tools.map((def) => toTool(name, entry, def))
|
|
|
+ entry.prompts = []
|
|
|
+ entry.status = { status: "connected" }
|
|
|
+ watch(name, entry, result.value.connection)
|
|
|
+ yield* Effect.logInfo("mcp connected", { server: name, tools: entry.tools.length })
|
|
|
+ // Announce the new tool set so the tool registry registers it. A server that finishes connecting
|
|
|
+ // after the initial registration sweep and emits no list-changed notification would otherwise
|
|
|
+ // stay invisible to the model.
|
|
|
yield* bus.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore)
|
|
|
yield* bus.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore)
|
|
|
- yield* bus.publish(Command.Event.Updated, {}).pipe(Effect.ignore)
|
|
|
yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
|
|
|
- }),
|
|
|
- ),
|
|
|
- )
|
|
|
- connection.onLog((message) => fork(serverLog(name, message).pipe(Effect.ignore)))
|
|
|
- connection.onToolsChanged(() =>
|
|
|
- live(
|
|
|
- refreshTools(name, entry, connection).pipe(
|
|
|
- Effect.andThen(bus.publish(McpEvent.ToolsChanged, { server: name })),
|
|
|
- ),
|
|
|
- ),
|
|
|
- )
|
|
|
- connection.onPromptsChanged(() => live(refreshPrompts(name, entry, connection)))
|
|
|
- connection.onResourcesChanged(() => live(bus.publish(McpEvent.ResourcesChanged, { server: name })))
|
|
|
- }
|
|
|
+ whenLive(name, entry, result.value.connection)(refreshPrompts(name, entry, result.value.connection))
|
|
|
+ return
|
|
|
+ }
|
|
|
+ yield* Scope.close(scope, Exit.void)
|
|
|
+ entry.scope = undefined
|
|
|
+ const error = Cause.squash(result.cause)
|
|
|
+ entry.status =
|
|
|
+ error instanceof MCPClient.NeedsAuthError
|
|
|
+ ? { status: "needs_auth" }
|
|
|
+ : { status: "failed", error: error instanceof Error ? error.message : String(error) }
|
|
|
+ yield* Effect.logWarning("mcp connect failed", { server: name, status: entry.status })
|
|
|
+ yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
|
|
|
+ }).pipe(Effect.ensuring(Deferred.succeed(entry.startup, undefined)))
|
|
|
|
|
|
- const serverLog = (server: ServerName, message: MCPClient.LogMessage) => {
|
|
|
- const fields = { server, logger: message.logger, level: message.level, data: message.data }
|
|
|
- switch (message.level) {
|
|
|
- case "debug":
|
|
|
- return Effect.logDebug("MCP server log", fields)
|
|
|
- case "info":
|
|
|
- case "notice":
|
|
|
- return Effect.logInfo("MCP server log", fields)
|
|
|
- case "warning":
|
|
|
- return Effect.logWarning("MCP server log", fields)
|
|
|
- case "error":
|
|
|
- case "critical":
|
|
|
- case "alert":
|
|
|
- case "emergency":
|
|
|
- return Effect.logError("MCP server log", fields)
|
|
|
- }
|
|
|
- }
|
|
|
+ const stopServer = Effect.fnUntraced(function* (name: ServerName, entry: ServerEntry) {
|
|
|
+ const scope = entry.scope
|
|
|
+ if (!scope) return
|
|
|
+ entry.scope = undefined
|
|
|
+ entry.client = undefined
|
|
|
+ entry.tools = undefined
|
|
|
+ entry.prompts = undefined
|
|
|
+ yield* Scope.close(scope, Exit.void)
|
|
|
+ yield* bus.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore)
|
|
|
+ yield* bus.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore)
|
|
|
+ yield* bus.publish(Command.Event.Updated, {}).pipe(Effect.ignore)
|
|
|
+ })
|
|
|
|
|
|
- const startServer = (name: ServerName, entry: ServerEntry) =>
|
|
|
- Effect.gen(function* () {
|
|
|
- // Announce the handshake so connect() and credential reconnects don't show a stale
|
|
|
- // disabled/failed status for the duration of the connection attempt.
|
|
|
- entry.status = { status: "pending" }
|
|
|
- yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
|
|
|
- const scope = yield* Scope.fork(root)
|
|
|
- entry.scope = scope
|
|
|
- const authProvider = yield* connectProvider(entry)
|
|
|
- // List tools as part of connect so a failure here marks the server failed rather than
|
|
|
- // leaving it connected with a silently empty tool list and no path to recover.
|
|
|
- const result = yield* MCPClient.connect(
|
|
|
- name,
|
|
|
- entry.config,
|
|
|
- location.directory,
|
|
|
- authProvider,
|
|
|
- elicitation,
|
|
|
- options?.clientInfo,
|
|
|
- ).pipe(
|
|
|
- Effect.flatMap((connection) => connection.tools().pipe(Effect.map((tools) => ({ connection, tools })))),
|
|
|
- Scope.provide(scope),
|
|
|
- Effect.exit,
|
|
|
- )
|
|
|
- if (Exit.isSuccess(result)) {
|
|
|
- entry.client = result.value.connection
|
|
|
- entry.tools = result.value.tools.map((def) => toTool(name, entry, def))
|
|
|
- entry.prompts = []
|
|
|
- entry.status = { status: "connected" }
|
|
|
- watch(name, entry, result.value.connection)
|
|
|
- yield* Effect.logInfo("mcp connected", { server: name, tools: entry.tools.length })
|
|
|
- // Announce the new tool set so the tool registry registers it. A server that finishes connecting
|
|
|
- // after the initial registration sweep and emits no list-changed notification would otherwise
|
|
|
- // stay invisible to the model.
|
|
|
- yield* bus.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore)
|
|
|
- yield* bus.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore)
|
|
|
- yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
|
|
|
- whenLive(name, entry, result.value.connection)(refreshPrompts(name, entry, result.value.connection))
|
|
|
- return
|
|
|
+ // Disabled servers settle their startup immediately so queries never block on them.
|
|
|
+ for (const [name, entry] of runtime) {
|
|
|
+ if (entry.config.disabled) {
|
|
|
+ entry.status = { status: "disabled" }
|
|
|
+ Deferred.doneUnsafe(entry.startup, Exit.void)
|
|
|
+ continue
|
|
|
}
|
|
|
- yield* Scope.close(scope, Exit.void)
|
|
|
- entry.scope = undefined
|
|
|
- const error = Cause.squash(result.cause)
|
|
|
- entry.status =
|
|
|
- error instanceof MCPClient.NeedsAuthError
|
|
|
- ? { status: "needs_auth" }
|
|
|
- : { status: "failed", error: error instanceof Error ? error.message : String(error) }
|
|
|
- yield* Effect.logWarning("mcp connect failed", { server: name, status: entry.status })
|
|
|
- yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
|
|
|
- }).pipe(Effect.ensuring(Deferred.succeed(entry.startup, undefined)))
|
|
|
-
|
|
|
- const stopServer = Effect.fnUntraced(function* (name: ServerName, entry: ServerEntry) {
|
|
|
- const scope = entry.scope
|
|
|
- if (!scope) return
|
|
|
- entry.scope = undefined
|
|
|
- entry.client = undefined
|
|
|
- entry.tools = undefined
|
|
|
- entry.prompts = undefined
|
|
|
- yield* Scope.close(scope, Exit.void)
|
|
|
- yield* bus.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore)
|
|
|
- yield* bus.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore)
|
|
|
- yield* bus.publish(Command.Event.Updated, {}).pipe(Effect.ignore)
|
|
|
- })
|
|
|
-
|
|
|
- // Disabled servers settle their startup immediately so queries never block on them.
|
|
|
- for (const [name, entry] of runtime) {
|
|
|
- if (entry.config.disabled) {
|
|
|
- entry.status = { status: "disabled" }
|
|
|
- Deferred.doneUnsafe(entry.startup, Exit.void)
|
|
|
- continue
|
|
|
+ fork(startServer(name, entry).pipe(locks.withLock(name)))
|
|
|
}
|
|
|
- fork(startServer(name, entry).pipe(locks.withLock(name)))
|
|
|
- }
|
|
|
|
|
|
- // Bring a server online (or back to needs_auth) when its integration's credential changes, so an
|
|
|
- // OAuth login takes effect without a restart. Only fires for the integrations we registered.
|
|
|
- const reconnect = (integrationID: Integration.ID) =>
|
|
|
- Effect.gen(function* () {
|
|
|
- const match = Array.from(runtime).find(([, entry]) => entry.integrationID === integrationID)
|
|
|
- if (!match) return
|
|
|
- const name = match[0]
|
|
|
- yield* Effect.gen(function* () {
|
|
|
- // add() or remove() may have replaced or deleted the entry while we waited for the lock.
|
|
|
- const entry = runtime.get(name)
|
|
|
- if (!entry || entry.integrationID !== integrationID) return
|
|
|
- if (entry.status.status === "disabled") return
|
|
|
- yield* stopServer(name, entry)
|
|
|
- yield* startServer(name, entry)
|
|
|
- }).pipe(locks.withLock(name))
|
|
|
- })
|
|
|
- fork(
|
|
|
- bus.subscribe(Integration.Event.ConnectionUpdated).pipe(
|
|
|
- Stream.filter((event) => owned.has(event.data.integrationID)),
|
|
|
- Stream.runForEach((event) => Effect.sync(() => fork(reconnect(event.data.integrationID)))),
|
|
|
- Effect.ignore,
|
|
|
- ),
|
|
|
- )
|
|
|
-
|
|
|
- // Suspend so each await sees current entries; a bare Map iterator is exhausted after one run.
|
|
|
- const whenAllReady = Effect.suspend(() =>
|
|
|
- Effect.forEach(Array.from(runtime.values()), (entry) => Deferred.await(entry.startup), {
|
|
|
- concurrency: "unbounded",
|
|
|
- discard: true,
|
|
|
- }),
|
|
|
- )
|
|
|
- return Service.of({
|
|
|
- servers: Effect.fn("MCP.servers")(function* () {
|
|
|
- return Array.from(runtime)
|
|
|
- .toSorted(([a], [b]) => a.localeCompare(b))
|
|
|
- .map(
|
|
|
- ([name, entry]) =>
|
|
|
- new ServerInfo({ name, status: entry.status, integrationID: entry.integrationID }),
|
|
|
- )
|
|
|
- }),
|
|
|
- add: Effect.fn("MCP.add")(function* (server, config) {
|
|
|
- const name = ServerName.make(server)
|
|
|
- yield* Effect.gen(function* () {
|
|
|
- const previous = runtime.get(name)
|
|
|
- if (previous) {
|
|
|
- yield* stopServer(name, previous)
|
|
|
- if (previous.integrationID) owned.delete(previous.integrationID)
|
|
|
- if (previous.registration) yield* previous.registration.dispose
|
|
|
- }
|
|
|
- const entry: ServerEntry = {
|
|
|
- config: { ...config, timeout: { ...timeout, ...config.timeout } },
|
|
|
- status: { status: "pending" },
|
|
|
- startup: Deferred.makeUnsafe<void>(),
|
|
|
- }
|
|
|
- runtime.set(name, entry)
|
|
|
+ // Bring a server online (or back to needs_auth) when its integration's credential changes, so an
|
|
|
+ // OAuth login takes effect without a restart. Only fires for the integrations we registered.
|
|
|
+ const reconnect = (integrationID: Integration.ID) =>
|
|
|
+ Effect.gen(function* () {
|
|
|
+ const match = Array.from(runtime).find(([, entry]) => entry.integrationID === integrationID)
|
|
|
+ if (!match) return
|
|
|
+ const name = match[0]
|
|
|
yield* Effect.gen(function* () {
|
|
|
- yield* register(name, entry)
|
|
|
- if (config.disabled) {
|
|
|
- entry.status = { status: "disabled" }
|
|
|
- yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
|
|
|
- return
|
|
|
- }
|
|
|
+ // add() or remove() may have replaced or deleted the entry while we waited for the lock.
|
|
|
+ const entry = runtime.get(name)
|
|
|
+ if (!entry || entry.integrationID !== integrationID) return
|
|
|
+ if (entry.status.status === "disabled") return
|
|
|
+ yield* stopServer(name, entry)
|
|
|
yield* startServer(name, entry)
|
|
|
- }).pipe(
|
|
|
- // Settle startup even when register fails or add is interrupted, so an entry that made it
|
|
|
- // into runtime can never hang readers awaiting its startup.
|
|
|
- Effect.ensuring(Effect.sync(() => Deferred.doneUnsafe(entry.startup, Exit.void))),
|
|
|
- )
|
|
|
- }).pipe(locks.withLock(name))
|
|
|
- }),
|
|
|
- connect: Effect.fn("MCP.connect")(function* (server) {
|
|
|
- const name = ServerName.make(server)
|
|
|
- yield* Effect.gen(function* () {
|
|
|
- const target = yield* requireServer(name)
|
|
|
- yield* stopServer(name, target.entry)
|
|
|
- yield* startServer(name, target.entry)
|
|
|
- }).pipe(locks.withLock(name))
|
|
|
- }),
|
|
|
- disconnect: Effect.fn("MCP.disconnect")(function* (server) {
|
|
|
- const name = ServerName.make(server)
|
|
|
- yield* Effect.gen(function* () {
|
|
|
- const target = yield* requireServer(name)
|
|
|
- yield* stopServer(name, target.entry)
|
|
|
- target.entry.status = { status: "disabled" }
|
|
|
- yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
|
|
|
- }).pipe(locks.withLock(name))
|
|
|
- }),
|
|
|
- remove: Effect.fn("MCP.remove")(function* (server) {
|
|
|
- const name = ServerName.make(server)
|
|
|
- yield* Effect.gen(function* () {
|
|
|
- const target = yield* requireServer(name)
|
|
|
- yield* stopServer(name, target.entry)
|
|
|
- if (target.entry.integrationID) owned.delete(target.entry.integrationID)
|
|
|
- if (target.entry.registration) yield* target.entry.registration.dispose
|
|
|
- // Credentials are kept: they are keyed by name + url, so re-adding the same server
|
|
|
- // reuses them without forcing re-auth, matching add()'s replacement semantics.
|
|
|
- runtime.delete(name)
|
|
|
- yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
|
|
|
- }).pipe(locks.withLock(name))
|
|
|
- }),
|
|
|
- tools: Effect.fn("MCP.tools")(function* () {
|
|
|
- yield* whenAllReady
|
|
|
- return Array.from(runtime.values())
|
|
|
- .flatMap((entry) => entry.tools ?? [])
|
|
|
- .toSorted((a, b) => a.server.localeCompare(b.server) || a.name.localeCompare(b.name))
|
|
|
- }),
|
|
|
- callTool: Effect.fn("MCP.callTool")(function* (input) {
|
|
|
- const target = yield* requireServer(input.server)
|
|
|
- yield* Deferred.await(target.entry.startup)
|
|
|
- if (!target.entry.client)
|
|
|
- return yield* new ToolCallError({
|
|
|
+ }).pipe(locks.withLock(name))
|
|
|
+ })
|
|
|
+ fork(
|
|
|
+ bus.subscribe(Integration.Event.ConnectionUpdated).pipe(
|
|
|
+ Stream.filter((event) => owned.has(event.data.integrationID)),
|
|
|
+ Stream.runForEach((event) => Effect.sync(() => fork(reconnect(event.data.integrationID)))),
|
|
|
+ Effect.ignore,
|
|
|
+ ),
|
|
|
+ )
|
|
|
+
|
|
|
+ // Suspend so each await sees current entries; a bare Map iterator is exhausted after one run.
|
|
|
+ const whenAllReady = Effect.suspend(() =>
|
|
|
+ Effect.forEach(Array.from(runtime.values()), (entry) => Deferred.await(entry.startup), {
|
|
|
+ concurrency: "unbounded",
|
|
|
+ discard: true,
|
|
|
+ }),
|
|
|
+ )
|
|
|
+ return Service.of({
|
|
|
+ servers: Effect.fn("MCP.servers")(function* () {
|
|
|
+ return Array.from(runtime)
|
|
|
+ .toSorted(([a], [b]) => a.localeCompare(b))
|
|
|
+ .map(([name, entry]) => new ServerInfo({ name, status: entry.status, integrationID: entry.integrationID }))
|
|
|
+ }),
|
|
|
+ add: Effect.fn("MCP.add")(function* (server, config) {
|
|
|
+ const name = ServerName.make(server)
|
|
|
+ yield* Effect.gen(function* () {
|
|
|
+ const previous = runtime.get(name)
|
|
|
+ if (previous) {
|
|
|
+ yield* stopServer(name, previous)
|
|
|
+ if (previous.integrationID) owned.delete(previous.integrationID)
|
|
|
+ if (previous.registration) yield* previous.registration.dispose
|
|
|
+ }
|
|
|
+ const entry: ServerEntry = {
|
|
|
+ config: { ...config, timeout: { ...timeout, ...config.timeout } },
|
|
|
+ status: { status: "pending" },
|
|
|
+ startup: Deferred.makeUnsafe<void>(),
|
|
|
+ }
|
|
|
+ runtime.set(name, entry)
|
|
|
+ yield* Effect.gen(function* () {
|
|
|
+ yield* register(name, entry)
|
|
|
+ if (config.disabled) {
|
|
|
+ entry.status = { status: "disabled" }
|
|
|
+ yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
|
|
|
+ return
|
|
|
+ }
|
|
|
+ yield* startServer(name, entry)
|
|
|
+ }).pipe(
|
|
|
+ // Settle startup even when register fails or add is interrupted, so an entry that made it
|
|
|
+ // into runtime can never hang readers awaiting its startup.
|
|
|
+ Effect.ensuring(Effect.sync(() => Deferred.doneUnsafe(entry.startup, Exit.void))),
|
|
|
+ )
|
|
|
+ }).pipe(locks.withLock(name))
|
|
|
+ }),
|
|
|
+ connect: Effect.fn("MCP.connect")(function* (server) {
|
|
|
+ const name = ServerName.make(server)
|
|
|
+ yield* Effect.gen(function* () {
|
|
|
+ const target = yield* requireServer(name)
|
|
|
+ yield* stopServer(name, target.entry)
|
|
|
+ yield* startServer(name, target.entry)
|
|
|
+ }).pipe(locks.withLock(name))
|
|
|
+ }),
|
|
|
+ disconnect: Effect.fn("MCP.disconnect")(function* (server) {
|
|
|
+ const name = ServerName.make(server)
|
|
|
+ yield* Effect.gen(function* () {
|
|
|
+ const target = yield* requireServer(name)
|
|
|
+ yield* stopServer(name, target.entry)
|
|
|
+ target.entry.status = { status: "disabled" }
|
|
|
+ yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
|
|
|
+ }).pipe(locks.withLock(name))
|
|
|
+ }),
|
|
|
+ remove: Effect.fn("MCP.remove")(function* (server) {
|
|
|
+ const name = ServerName.make(server)
|
|
|
+ yield* Effect.gen(function* () {
|
|
|
+ const target = yield* requireServer(name)
|
|
|
+ yield* stopServer(name, target.entry)
|
|
|
+ if (target.entry.integrationID) owned.delete(target.entry.integrationID)
|
|
|
+ if (target.entry.registration) yield* target.entry.registration.dispose
|
|
|
+ // Credentials are kept: they are keyed by name + url, so re-adding the same server
|
|
|
+ // reuses them without forcing re-auth, matching add()'s replacement semantics.
|
|
|
+ runtime.delete(name)
|
|
|
+ yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
|
|
|
+ }).pipe(locks.withLock(name))
|
|
|
+ }),
|
|
|
+ tools: Effect.fn("MCP.tools")(function* () {
|
|
|
+ yield* whenAllReady
|
|
|
+ return Array.from(runtime.values())
|
|
|
+ .flatMap((entry) => entry.tools ?? [])
|
|
|
+ .toSorted((a, b) => a.server.localeCompare(b.server) || a.name.localeCompare(b.name))
|
|
|
+ }),
|
|
|
+ callTool: Effect.fn("MCP.callTool")(function* (input) {
|
|
|
+ const target = yield* requireServer(input.server)
|
|
|
+ yield* Deferred.await(target.entry.startup)
|
|
|
+ if (!target.entry.client)
|
|
|
+ return yield* new ToolCallError({
|
|
|
+ server: target.name,
|
|
|
+ tool: input.name,
|
|
|
+ message: "MCP server is not connected",
|
|
|
+ })
|
|
|
+ const result = yield* target.entry.client
|
|
|
+ .callTool({ name: input.name, args: input.args })
|
|
|
+ .pipe(
|
|
|
+ Effect.mapError(
|
|
|
+ (error) => new ToolCallError({ server: target.name, tool: input.name, message: error.message }),
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ return new ToolResult({
|
|
|
server: target.name,
|
|
|
tool: input.name,
|
|
|
- message: "MCP server is not connected",
|
|
|
+ isError: result.isError,
|
|
|
+ structured: result.structured,
|
|
|
+ content: result.content,
|
|
|
})
|
|
|
- const result = yield* target.entry.client
|
|
|
- .callTool({ name: input.name, args: input.args })
|
|
|
- .pipe(
|
|
|
- Effect.mapError(
|
|
|
- (error) => new ToolCallError({ server: target.name, tool: input.name, message: error.message }),
|
|
|
+ }),
|
|
|
+ instructions: Effect.fn("MCP.instructions")(function* () {
|
|
|
+ yield* whenAllReady
|
|
|
+ return Array.from(runtime)
|
|
|
+ .flatMap(([server, entry]) => {
|
|
|
+ const instructions = entry.client?.instructions
|
|
|
+ if (!instructions) return []
|
|
|
+ return [new ServerInstructions({ server, instructions })]
|
|
|
+ })
|
|
|
+ .toSorted((a, b) => a.server.localeCompare(b.server))
|
|
|
+ }),
|
|
|
+ prompts: Effect.fn("MCP.prompts")(function* () {
|
|
|
+ return Array.from(runtime.values())
|
|
|
+ .flatMap((entry) => entry.prompts ?? [])
|
|
|
+ .toSorted((a, b) => a.server.localeCompare(b.server) || a.name.localeCompare(b.name))
|
|
|
+ }),
|
|
|
+ prompt: Effect.fn("MCP.prompt")(function* (input) {
|
|
|
+ const target = yield* requireServer(input.server)
|
|
|
+ yield* Deferred.await(target.entry.startup)
|
|
|
+ if (!target.entry.client) return undefined
|
|
|
+ const result = yield* target.entry.client
|
|
|
+ .prompt({ name: input.name, args: input.args })
|
|
|
+ .pipe(Effect.catch(() => Effect.succeed(undefined)))
|
|
|
+ if (!result) return undefined
|
|
|
+ return new PromptResult({
|
|
|
+ server: target.name,
|
|
|
+ name: input.name,
|
|
|
+ messages: result.messages.map(
|
|
|
+ (message) => new PromptMessage({ role: message.role, content: message.content }),
|
|
|
),
|
|
|
+ })
|
|
|
+ }),
|
|
|
+ resourceCatalog: Effect.fn("MCP.resourceCatalog")(function* () {
|
|
|
+ yield* whenAllReady
|
|
|
+ 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 new ToolResult({
|
|
|
- server: target.name,
|
|
|
- tool: input.name,
|
|
|
- isError: result.isError,
|
|
|
- structured: result.structured,
|
|
|
- content: result.content,
|
|
|
- })
|
|
|
- }),
|
|
|
- instructions: Effect.fn("MCP.instructions")(function* () {
|
|
|
- yield* whenAllReady
|
|
|
- return Array.from(runtime)
|
|
|
- .flatMap(([server, entry]) => {
|
|
|
- const instructions = entry.client?.instructions
|
|
|
- if (!instructions) return []
|
|
|
- return [new ServerInstructions({ server, instructions })]
|
|
|
+ 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),
|
|
|
+ ),
|
|
|
})
|
|
|
- .toSorted((a, b) => a.server.localeCompare(b.server))
|
|
|
- }),
|
|
|
- prompts: Effect.fn("MCP.prompts")(function* () {
|
|
|
- return Array.from(runtime.values())
|
|
|
- .flatMap((entry) => entry.prompts ?? [])
|
|
|
- .toSorted((a, b) => a.server.localeCompare(b.server) || a.name.localeCompare(b.name))
|
|
|
- }),
|
|
|
- prompt: Effect.fn("MCP.prompt")(function* (input) {
|
|
|
- const target = yield* requireServer(input.server)
|
|
|
- yield* Deferred.await(target.entry.startup)
|
|
|
- if (!target.entry.client) return undefined
|
|
|
- const result = yield* target.entry.client
|
|
|
- .prompt({ name: input.name, args: input.args })
|
|
|
- .pipe(Effect.catch(() => Effect.succeed(undefined)))
|
|
|
- if (!result) return undefined
|
|
|
- return new PromptResult({
|
|
|
- server: target.name,
|
|
|
- name: input.name,
|
|
|
- messages: result.messages.map(
|
|
|
- (message) => new PromptMessage({ role: message.role, content: message.content }),
|
|
|
- ),
|
|
|
- })
|
|
|
- }),
|
|
|
- resourceCatalog: Effect.fn("MCP.resourceCatalog")(function* () {
|
|
|
- yield* whenAllReady
|
|
|
- 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) {
|
|
|
- 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,
|
|
|
- })
|
|
|
- }),
|
|
|
- })
|
|
|
- }),
|
|
|
-)
|
|
|
+ }),
|
|
|
+ readResource: Effect.fn("MCP.readResource")(function* (input) {
|
|
|
+ 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,
|
|
|
+ })
|
|
|
+ }),
|
|
|
+ })
|
|
|
+ }),
|
|
|
+ )
|
|
|
|
|
|
export function configured(options?: Options) {
|
|
|
return makeLocationNode({
|