feat: add MCP tool conversion util #5003
Conversation
|
Awesome - left some comments. It would be great to have an example under |
|
@iteratetograceness actually would it be possible to call the MCP websocket api directly without their library? if so, how hard would it be? that would solve the dependency issues and allow us to include the MCP tool directly in the |
Thinking about this: seems like what we want is a custom Client and Transport; the trade-off here being we would essentially be copying over chunks of their code that we want to use |
I usually re-develop from scratch against APIs to make sure the code is up to our standards. |
| /** Optional client version, defaults to '1.0.0' */ | ||
| version?: string; |
There was a problem hiding this comment.
hm the client version is what we support, right? so should the user be able to specify it?
There was a problem hiding this comment.
client config is more informational, not really used in a critical capacity (e.g. compatibility checking b/w client and server) - i'd like to advocate for keeping it so that users can debug/track specific clients they create if necessary!
packages/ai/core/tool/mcp/client.ts
Outdated
|
|
||
| return this; | ||
| } catch (error) { | ||
| void this.close(); |
There was a problem hiding this comment.
void is not needed? also, what are the implications of calling close here?
There was a problem hiding this comment.
noted on void! calling close here ensures we clean up the transport/connection and clear our response handlers
packages/ai/core/tool/mcp/client.ts
Outdated
| const responseHandlers = this.responseHandlers; | ||
| this.responseHandlers = new Map(); | ||
|
|
||
| const error = new MCPClientError({ | ||
| message: 'Connection closed', | ||
| }); | ||
|
|
||
| for (const handler of responseHandlers.values()) { | ||
| handler(error); | ||
| } |
There was a problem hiding this comment.
would the actual connection need to be closed? also, why do we need to swap response handlers? can we set an isClosed flag on the object instead?
also, i prefer onError, onClose
There was a problem hiding this comment.
yes! we invoke this.transport?.close() when this.onclose is called!
packages/ai/core/tool/mcp/client.ts
Outdated
| private onerror(error: Error): void { | ||
| throw new MCPClientError({ | ||
| message: error.message, | ||
| cause: error, | ||
| }); | ||
| } |
There was a problem hiding this comment.
what will this do in practice (in terms of the stack and where the error shows up)?
There was a problem hiding this comment.
ah, good point! since the client's onError (which is triggered via the transport's onerror handler or onResponse) we aren't properly propagating this up within the client
There was a problem hiding this comment.
instead of throwing MCPClientError, should we just close the client+transport? we need a way to properly bubble up the error to the user hmmmmm
|
|
||
| let buffer = ''; | ||
|
|
||
| const processEvents = async () => { |
There was a problem hiding this comment.
can we use the eventstream parser package that we are already using in provider utils?
|
|
||
| buffer += textDecoder.decode(value, { stream: true }); | ||
| const events = buffer.split('\n\n'); | ||
| buffer = events.pop() || ''; |
There was a problem hiding this comment.
added in case we get a chunk of that includes the beginning of the next event!
| const DEFAULT_INHERITED_ENV_VARS = | ||
| process.platform === 'win32' | ||
| ? [ | ||
| 'APPDATA', | ||
| 'HOMEDRIVE', | ||
| 'HOMEPATH', | ||
| 'LOCALAPPDATA', | ||
| 'PATH', | ||
| 'PROCESSOR_ARCHITECTURE', | ||
| 'SYSTEMDRIVE', | ||
| 'SYSTEMROOT', | ||
| 'TEMP', | ||
| 'USERNAME', | ||
| 'USERPROFILE', | ||
| ] | ||
| : ['HOME', 'LOGNAME', 'PATH', 'SHELL', 'TERM', 'USER']; | ||
|
|
||
| function getDefaultEnvironment(): Record<string, string> { | ||
| const env: Record<string, string> = {}; | ||
|
|
||
| for (const key of DEFAULT_INHERITED_ENV_VARS) { | ||
| const value = process.env[key]; | ||
| if (value === undefined) { | ||
| continue; | ||
| } | ||
|
|
||
| if (value.startsWith('()')) { | ||
| continue; | ||
| } | ||
|
|
||
| env[key] = value; | ||
| } | ||
|
|
||
| return env; | ||
| } | ||
|
|
||
| function isElectron() { | ||
| return 'type' in process; | ||
| } |
There was a problem hiding this comment.
used for config when spawning child_process!
| const RequestIdSchema = z.union([z.string(), z.number().int()]); | ||
| const JSONRPCRequestSchema = z | ||
| .object({ | ||
| jsonrpc: z.literal(JSONRPC_VERSION), | ||
| id: RequestIdSchema, | ||
| }) | ||
| .merge(RequestSchema) | ||
| .strict(); | ||
| export type JSONRPCRequest = z.infer<typeof JSONRPCRequestSchema>; | ||
| const JSONRPCResponseSchema = z | ||
| .object({ | ||
| jsonrpc: z.literal(JSONRPC_VERSION), | ||
| id: RequestIdSchema, | ||
| result: ResultSchema, | ||
| }) | ||
| .strict(); | ||
| export type JSONRPCResponse = z.infer<typeof JSONRPCResponseSchema>; | ||
| const JSONRPCErrorSchema = z | ||
| .object({ | ||
| jsonrpc: z.literal(JSONRPC_VERSION), | ||
| id: RequestIdSchema, | ||
| error: z.object({ | ||
| code: z.number().int(), | ||
| message: z.string(), | ||
| data: z.optional(z.unknown()), | ||
| }), | ||
| }) | ||
| .strict(); | ||
| export type JSONRPCError = z.infer<typeof JSONRPCErrorSchema>; | ||
| const JSONRPCNotificationSchema = z | ||
| .object({ | ||
| jsonrpc: z.literal(JSONRPC_VERSION), | ||
| }) | ||
| .merge(NotificationSchema) | ||
| .strict(); | ||
| export type JSONRPCNotification = z.infer<typeof JSONRPCNotificationSchema>; | ||
|
|
||
| export const JSONRPCMessageSchema = z.union([ | ||
| JSONRPCRequestSchema, | ||
| JSONRPCNotificationSchema, | ||
| JSONRPCResponseSchema, | ||
| JSONRPCErrorSchema, | ||
| ]); | ||
| export type JSONRPCMessage = z.infer<typeof JSONRPCMessageSchema>; | ||
|
|
||
| export interface MCPTransport { | ||
| start(): Promise<void>; | ||
| send(message: JSONRPCMessage): Promise<void>; |
There was a problem hiding this comment.
do we need zod schemas for all of this (as in, we use the schemas), or can we go more lightweight and just define TS types (at least for some)?
| @@ -104,6 +122,7 @@ export function createTestServer<URLS extends { [url: string]: UrlHandler }>( | |||
| ): { | |||
| urls: FullHandlers<URLS>; | |||
| calls: TestServerCall[]; | |||
| addSseEvent: (url: string, data: any, event?: string, id?: string) => void; | |||
There was a problem hiding this comment.
wondering about this because it is sse specific. there is another pattern i used, see e.g. https://github.com/vercel/ai/blob/main/packages/react/src/use-chat.ui.test.tsx#L880
wondering if that would be an alternative (maybe it's the same thing and addServerSentEvent or addSSE (rename?) is more elegant
There was a problem hiding this comment.
also do we want to be SSE specific here? this could just a generic stream and you could attach SSE formatted chunks (we could also expose formatSSEChunk as a separate helper)
There was a problem hiding this comment.
my approach is missing a sort of streamController like your approach is 👀 hoping to discuss in call tomo!
No description provided.