diff --git a/dist/unstable/ai/McpServer.js b/dist/unstable/ai/McpServer.js index 755e41c..dc431aa 100644 --- a/dist/unstable/ai/McpServer.js +++ b/dist/unstable/ai/McpServer.js @@ -262,6 +262,26 @@ const runWithProtocolState = /*#__PURE__*/Effect.fnUntraced(function* (options, const server = yield* McpServer; const isHttp = Option.isSome(yield* Effect.serviceOption(HttpRouter.HttpRouter)); const sessions = protocolState.sessions; + if (isHttp && options.path !== undefined) { + const router = yield* HttpRouter.HttpRouter; + yield* router.add("DELETE", options.path, Effect.gen(function* () { + const request = yield* HttpServerRequest.HttpServerRequest; + const sessionId = request.headers[MCP_SESSION_ID_HEADER]; + if (sessionId === undefined) { + return HttpServerResponse.empty({ + status: 400 + }); + } + if (!sessions.bySessionId.delete(sessionId)) { + return HttpServerResponse.empty({ + status: 404 + }); + } + return HttpServerResponse.empty({ + status: 204 + }); + })); + } const clientProtocols = new Map(); const cancelledRequests = new Map(); const handlers = yield* Layer.build(layerHandlers(options, { @@ -719,7 +739,7 @@ export const layerHttp = options => { const methodNotAllowed = request => Effect.succeed(request.headers.origin !== undefined && !options.allowedOrigins?.includes(request.headers.origin) ? HttpServerResponse.empty({ status: 403 }) : methodNotAllowedResponse); - const routes = Layer.mergeAll(HttpRouter.add("GET", options.path, methodNotAllowed), HttpRouter.add("PUT", options.path, methodNotAllowed), HttpRouter.add("PATCH", options.path, methodNotAllowed), HttpRouter.add("DELETE", options.path, methodNotAllowed), HttpRouter.add("OPTIONS", options.path, methodNotAllowed)); + const routes = Layer.mergeAll(HttpRouter.add("GET", options.path, methodNotAllowed), HttpRouter.add("PUT", options.path, methodNotAllowed), HttpRouter.add("PATCH", options.path, methodNotAllowed), HttpRouter.add("OPTIONS", options.path, methodNotAllowed)); return Layer.merge(layerWithProtocolState(options), routes).pipe(Layer.provide(layerMcpProtocolHttp(options)), Layer.provide(protocolState), Layer.provide(RpcSerialization.layerJsonRpc())); }; const layerMcpProtocolHttp = options => Layer.effect(RpcServer.Protocol)(Effect.gen(function* () { diff --git a/dist/unstable/rpc/RpcClient.d.ts b/dist/unstable/rpc/RpcClient.d.ts index aba2779..581d267 100644 --- a/dist/unstable/rpc/RpcClient.d.ts +++ b/dist/unstable/rpc/RpcClient.d.ts @@ -2,6 +2,7 @@ import * as Cause from "../../Cause.ts"; import * as Context from "../../Context.ts"; import type * as Duration from "../../Duration.ts"; import * as Effect from "../../Effect.ts"; +import * as Exit from "../../Exit.ts"; import * as Layer from "../../Layer.ts"; import * as Queue from "../../Queue.ts"; import * as Schedule from "../../Schedule.ts"; @@ -261,9 +262,40 @@ export declare const layerProtocolWorker: (options: { readonly targetUtilization?: number | undefined; readonly timeToLive: Duration.Input; }) => Layer.Layer; +declare const RequestHooks_base: Context.ServiceClass Effect.Effect) | undefined; + readonly onRequestChunk?: ((info: { + readonly id: RequestId; + readonly tag: string; + readonly chunkCount: number; + }) => Effect.Effect) | undefined; + readonly onRequestExit?: ((info: { + readonly id: RequestId; + readonly tag: string; + readonly stream: boolean; + readonly exit: Exit.Exit; + }) => Effect.Effect) | undefined; + readonly onRequestInterrupt?: ((info: { + readonly id: RequestId; + readonly tag?: string | undefined; + }) => Effect.Effect) | undefined; +}>; +/** + * @since 4.0.0 + * @category RequestHooks + */ +export declare class RequestHooks extends RequestHooks_base { +} declare const ConnectionHooks_base: Context.ServiceClass; readonly onDisconnect: Effect.Effect; + readonly onPing?: Effect.Effect | undefined; + readonly onPong?: Effect.Effect | undefined; + readonly onPingTimeout?: Effect.Effect | undefined; }>; /** * Represents optional client protocol hooks that run when a transport connects diff --git a/dist/unstable/rpc/RpcClient.js b/dist/unstable/rpc/RpcClient.js index b536d0a..12ffac0 100644 --- a/dist/unstable/rpc/RpcClient.js +++ b/dist/unstable/rpc/RpcClient.js @@ -47,6 +47,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro const services = yield* Effect.context(); const scope = Context.get(services, Scope.Scope); const entries = new Map(); + const hooks = yield* Effect.serviceOption(RequestHooks); let isShutdown = false; yield* Scope.addFinalizer(scope, Effect.withFiber(parent => { isShutdown = true; @@ -84,6 +85,11 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro return Effect.interrupt; } const id = generateRequestId(); + const onStart = Option.isSome(hooks) && hooks.value.onRequestStart ? hooks.value.onRequestStart({ + id, + tag: rpc._tag, + stream: false + }) : Effect.void; const send = middleware(message => options.onFromClient({ message, context, @@ -101,7 +107,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro headers: Headers.merge(parentFiber.getRef(CurrentHeaders), headers) }); if (discard) { - return send; + return onStart.pipe(Effect.andThen(send)); } let fiber; return Effect.onInterrupt(Effect.callback(resume => { @@ -119,7 +125,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro } }; entries.set(id, entry); - fiber = send.pipe(span ? Effect.withParentSpan(span, { + fiber = onStart.pipe(Effect.andThen(send), span ? Effect.withParentSpan(span, { captureStackTrace: false }) : identity, Effect.runForkWith(parentFiber.context)); fiber.addObserver(exit => { @@ -128,8 +134,9 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro } }); }), interruptors => { + const entry = entries.get(id); entries.delete(id); - return Effect.andThen(Fiber.interrupt(fiber), sendInterrupt(id, Array.from(interruptors), context)); + return Effect.andThen(Fiber.interrupt(fiber), sendInterrupt(id, Array.from(interruptors), context, entry?.rpc._tag)); }); }); const onStreamRequest = Effect.fnUntraced(function* (rpc, middleware, payload, headers, streamBufferSize, context) { @@ -141,11 +148,17 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro }); const fiber = Fiber.getCurrent(); const id = generateRequestId(); + const onStart = Option.isSome(hooks) && hooks.value.onRequestStart ? hooks.value.onRequestStart({ + id, + tag: rpc._tag, + stream: true + }) : Effect.void; const scope = Context.getUnsafe(fiber.context, Scope.Scope); yield* Scope.addFinalizerExit(scope, exit => { - if (!entries.has(id)) return Effect.void; + const entry = entries.get(id); + if (!entry) return Effect.void; entries.delete(id); - return sendInterrupt(id, Exit.isFailure(exit) ? Array.from(Cause.interruptors(exit.cause)) : [], context); + return sendInterrupt(id, Exit.isFailure(exit) ? Array.from(Cause.interruptors(exit.cause)) : [], context, entry.rpc._tag); }); const queue = yield* Queue.bounded(streamBufferSize); entries.set(id, { @@ -153,9 +166,10 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro rpc, queue, scope, - context + context, + chunkCount: 0 }); - yield* middleware(message => options.onFromClient({ + yield* onStart.pipe(Effect.andThen(middleware(message => options.onFromClient({ message, context, discard: false @@ -170,7 +184,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro sampled: span.sampled } : {}), headers: Headers.merge(fiber.getRef(CurrentHeaders), headers) - }).pipe(span ? Effect.withParentSpan(span, { + }))).pipe(span ? Effect.withParentSpan(span, { captureStackTrace: false }) : identity, Effect.catchCause(error => Queue.failCause(queue, error)), Effect.interruptible, Effect.forkIn(scope, { startImmediately: true @@ -200,7 +214,10 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro }); }; }; - const sendInterrupt = (requestId, interruptors, context) => Effect.callback(resume => { + const sendInterrupt = (requestId, interruptors, context, tag) => (Option.isSome(hooks) && hooks.value.onRequestInterrupt ? hooks.value.onRequestInterrupt({ + id: requestId, + tag + }) : Effect.void).pipe(Effect.andThen(Effect.callback(resume => { const parentFiber = Fiber.getCurrent(); const fiber = options.onFromClient({ message: { @@ -214,7 +231,7 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro fiber.addObserver(() => { resume(Effect.void); }); - }); + }))); const write = message => { switch (message._tag) { case "Chunk": @@ -222,7 +239,13 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro const requestId = message.requestId; const entry = entries.get(requestId); if (!entry || entry._tag !== "Queue") return Effect.void; - return Queue.offerAll(entry.queue, message.values).pipe(supportsAck ? Effect.flatMap(() => options.onFromClient({ + const chunkCount = entry.chunkCount += message.values.length; + const onChunk = Option.isSome(hooks) && hooks.value.onRequestChunk ? hooks.value.onRequestChunk({ + id: requestId, + tag: entry.rpc._tag, + chunkCount + }) : Effect.void; + return Queue.offerAll(entry.queue, message.values).pipe(Effect.andThen(onChunk), supportsAck ? Effect.flatMap(() => options.onFromClient({ message: { _tag: "Ack", requestId: message.requestId @@ -237,11 +260,18 @@ export const makeNoSerialization = /*#__PURE__*/Effect.fnUntraced(function* (gro const entry = entries.get(requestId); if (!entry) return Effect.void; entries.delete(requestId); + const onExit = Option.isSome(hooks) && hooks.value.onRequestExit ? hooks.value.onRequestExit({ + id: requestId, + tag: entry.rpc._tag, + stream: entry._tag === "Queue", + exit: message.exit + }) : Effect.void; if (entry._tag === "Effect") { - entry.resume(message.exit); - return Effect.void; + return onExit.pipe(Effect.andThen(Effect.sync(() => { + entry.resume(message.exit); + }))); } - return message.exit._tag === "Success" ? Queue.end(entry.queue) : Queue.failCause(entry.queue, message.exit.cause); + return onExit.pipe(Effect.andThen(message.exit._tag === "Success" ? Queue.end(entry.queue) : Queue.failCause(entry.queue, message.exit.cause))); } case "Defect": { @@ -590,7 +620,7 @@ export const makeProtocolSocket = options => Protocol.make(Effect.fnUntraced(fun const requestClientMap = new Map(); const write = yield* socket.writer; let parser = serialization.makeUnsafe(); - const pinger = yield* makePinger(write(parser.encode(constPing))); + const pinger = yield* makePinger(write(parser.encode(constPing)), Option.isSome(hooks) ? hooks.value : undefined); let currentError; const onOpen = Effect.suspend(() => { currentError = undefined; @@ -610,8 +640,7 @@ export const makeProtocolSocket = options => Protocol.make(Effect.fnUntraced(fun body: () => { const response = responses[i++]; if (response._tag === "Pong") { - pinger.onPong(); - return Effect.void; + return pinger.onPong; } if (Object.hasOwn(response, "requestId")) { const requestId = response.requestId; @@ -640,12 +669,12 @@ export const makeProtocolSocket = options => Protocol.make(Effect.fnUntraced(fun } }, { onOpen - }).pipe(Effect.raceFirst(Effect.flatMap(pinger.timeout, () => Effect.fail(new Socket.SocketError({ + }).pipe(Effect.raceFirst(Effect.flatMap(pinger.timeout, () => (Option.isSome(hooks) && hooks.value.onPingTimeout ? hooks.value.onPingTimeout : Effect.void).pipe(Effect.andThen(Effect.fail(new Socket.SocketError({ reason: new Socket.SocketOpenError({ kind: "Timeout", cause: new Error("ping timeout") }) - }))))); + }))))))); }).pipe(Effect.flatMap(() => Effect.fail(new Socket.SocketError({ reason: new Socket.SocketCloseError({ code: 1000 @@ -687,20 +716,28 @@ export const makeProtocolSocket = options => Protocol.make(Effect.fnUntraced(fun }; })); const defaultRetryPolicy = /*#__PURE__*/Schedule.min([/*#__PURE__*/Schedule.exponential(500, 1.5), /*#__PURE__*/Schedule.spaced(5000)]); -const makePinger = /*#__PURE__*/Effect.fnUntraced(function* (writePing) { +const makePinger = /*#__PURE__*/Effect.fnUntraced(function* (writePing, hooks) { let recievedPong = true; + let missedPongs = 0; const latch = Latch.makeUnsafe(); const reset = () => { recievedPong = true; + missedPongs = 0; latch.closeUnsafe(); }; - const onPong = () => { - recievedPong = true; - }; + const onPong = Effect.sync(() => { + recievedPong = true; + missedPongs = 0; + }).pipe(Effect.andThen(hooks?.onPong ?? Effect.void)); yield* Effect.suspend(() => { - if (!recievedPong) return latch.open; - recievedPong = false; - return writePing; + if (!recievedPong) { + missedPongs += 1; + if (missedPongs >= 3) return latch.open; + return (hooks?.onPing ?? Effect.void).pipe(Effect.andThen(writePing)); + } + recievedPong = false; + missedPongs = 0; + return (hooks?.onPing ?? Effect.void).pipe(Effect.andThen(writePing)); }).pipe(Effect.delay("5 seconds"), Effect.ignore, Effect.forever, Effect.interruptible, Effect.forkScoped); return { timeout: latch.await, @@ -843,6 +880,11 @@ export const makeProtocolWorker = options => Protocol.make(Effect.fnUntraced(fun * @since 4.0.0 */ export const layerProtocolWorker = /*#__PURE__*/flow(makeProtocolWorker, /*#__PURE__*/Layer.effect(Protocol)); +/** + * @since 4.0.0 + * @category RequestHooks + */ +export class RequestHooks extends /*#__PURE__*/Context.Service()("effect/rpc/RpcClient/RequestHooks") {} /** * Represents optional client protocol hooks that run when a transport connects * and disconnects.