mirror of
https://github.com/hanzoai/code.git
synced 2026-09-30 20:53:35 +00:00
Updated the channel it include events for add, update, delete, and loading. Fixed a bug related to client.close() throwing and error
This commit is contained in:
@@ -13,13 +13,14 @@ 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 { SSEClientTransport } from '@modelcontextprotocol/sdk/client/sse.js';
|
||||
import { MCPConfig, MCPServerConfig, EventMCPServerSetupOnError, EventMCPServerSetupOnSuccess, MCPServerSuccessModel, MCPServerErrorModel } from '../common/mcpServiceTypes.js';
|
||||
import { MCPConfig, MCPServerConfig, MCPServerErrorModel, MCPAddResponse, MCPServerEventAddParam, MCPServerEventUpdateParam, MCPServerEventDeleteParam, MCPUpdateResponse, MCPServerModel, MCPDeleteResponse, MCPServerEventLoadingParam } from '../common/mcpServiceTypes.js';
|
||||
import { Transport } from '@modelcontextprotocol/sdk/shared/transport.js';
|
||||
import { equals } from '../../../../base/common/objects.js';
|
||||
|
||||
export class MCPChannel implements IServerChannel {
|
||||
|
||||
// connected clients
|
||||
private clients: { [clientId: string]: { client: Client, mcpConfig: MCPServerConfig } } = {}
|
||||
private clients: { [clientId: string]: { client?: Client, mcpConfig: MCPServerConfig, formattedServer: MCPServerModel } } = {}
|
||||
private getClientConfig(serverName: string) {
|
||||
return {
|
||||
name: `${serverName}-client`,
|
||||
@@ -30,18 +31,22 @@ export class MCPChannel implements IServerChannel {
|
||||
|
||||
// mcp emitters
|
||||
private readonly mcpEmitters = {
|
||||
serverSetup: {
|
||||
success: new Emitter<EventMCPServerSetupOnSuccess<MCPServerSuccessModel>>(),
|
||||
error: new Emitter<EventMCPServerSetupOnError<MCPServerErrorModel>>(),
|
||||
},
|
||||
serverEvent: {
|
||||
add: new Emitter<MCPServerEventAddParam>(),
|
||||
update: new Emitter<MCPServerEventUpdateParam>(),
|
||||
delete: new Emitter<MCPServerEventDeleteParam>(),
|
||||
loading: new Emitter<MCPServerEventLoadingParam>(),
|
||||
}
|
||||
// toolCall: {
|
||||
// success: new Emitter<void>(),
|
||||
// error: new Emitter<void>(),
|
||||
// },
|
||||
} satisfies {
|
||||
[task in 'serverSetup']: {
|
||||
success: Emitter<EventMCPServerSetupOnSuccess<any>>,
|
||||
error: Emitter<EventMCPServerSetupOnError<any>>,
|
||||
[event in 'serverEvent']: {
|
||||
add: Emitter<MCPServerEventAddParam>,
|
||||
update: Emitter<MCPServerEventUpdateParam>,
|
||||
delete: Emitter<MCPServerEventDeleteParam>,
|
||||
loading: Emitter<MCPServerEventLoadingParam>,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -52,13 +57,11 @@ export class MCPChannel implements IServerChannel {
|
||||
// browser uses this to listen for changes
|
||||
listen(_: unknown, event: string): Event<any> {
|
||||
|
||||
// server setup
|
||||
if (event === 'onSuccess_serverSetup') return this.mcpEmitters.serverSetup.success.event;
|
||||
else if (event === 'onError_serverSetup') return this.mcpEmitters.serverSetup.error.event;
|
||||
|
||||
// tool call
|
||||
// else if (event === 'onSuccess_toolCall') return this.mcpEmitters.toolCall.success.event;
|
||||
// else if (event === 'onError_toolCall') return this.mcpEmitters.toolCall.error.event;
|
||||
// server events
|
||||
if (event === 'onAdd_server') return this.mcpEmitters.serverEvent.add.event;
|
||||
else if (event === 'onUpdate_server') return this.mcpEmitters.serverEvent.update.event;
|
||||
else if (event === 'onDelete_server') return this.mcpEmitters.serverEvent.delete.event;
|
||||
else if (event === 'onLoading_server') return this.mcpEmitters.serverEvent.loading.event;
|
||||
|
||||
// handle unknown events
|
||||
else throw new Error(`Event not found: ${event}`);
|
||||
@@ -92,10 +95,8 @@ export class MCPChannel implements IServerChannel {
|
||||
|
||||
private async _callSetupServers(mcpConfig: MCPConfig) {
|
||||
|
||||
// Reset all servers
|
||||
if (Object.keys(this.clients).length > 0) {
|
||||
await this._callCloseAllServers()
|
||||
}
|
||||
// Get all prevServers
|
||||
const prevServers = { ...this.clients }
|
||||
|
||||
// Handle config file setup and changes
|
||||
const { mcpServers } = mcpConfig
|
||||
@@ -105,42 +106,86 @@ export class MCPChannel implements IServerChannel {
|
||||
console.log('No MCP servers found in config file.')
|
||||
return
|
||||
}
|
||||
for (const serverName of serverNames) {
|
||||
const getPrevAndNewServerConfig = (serverName: string) => {
|
||||
const prevMCPConfig = prevServers[serverName]?.mcpConfig
|
||||
const newMCPConfig = mcpServers[serverName]
|
||||
return { prevMCPConfig, newMCPConfig }
|
||||
}
|
||||
|
||||
// Get the server config
|
||||
const server = mcpServers[serverName]
|
||||
|
||||
if (server) {
|
||||
// TODO: add a check if server is on or off
|
||||
try {
|
||||
await this._callSetupServer(server, serverName)
|
||||
|
||||
} catch (err) {
|
||||
// catches *any* error (including SSE fallback or Stdio connect)
|
||||
console.error(`❌ Failed to connect to server "${serverName}":`, err);
|
||||
// fire error event
|
||||
// TODO: handle sending back the error
|
||||
const typedErr = err as Error
|
||||
console.log('Error Message: ', typedErr.message)
|
||||
|
||||
// Get the full command string for display
|
||||
const fullCommand = `${server.command} ${server.args?.join(' ') || ''}`
|
||||
console.log('Full Command: ', fullCommand)
|
||||
|
||||
this.mcpEmitters.serverSetup.error.fire({
|
||||
model: {
|
||||
serverName,
|
||||
status: 'error',
|
||||
isOn: false,
|
||||
tools: [],
|
||||
error: typedErr.message,
|
||||
command: fullCommand,
|
||||
}
|
||||
})
|
||||
// and then move on to the next server
|
||||
continue;
|
||||
}
|
||||
// Divide the server based on event
|
||||
const addedServers = serverNames.filter((serverName) => {
|
||||
const { prevMCPConfig, newMCPConfig } = getPrevAndNewServerConfig(serverName)
|
||||
const isAdded = !prevMCPConfig && newMCPConfig
|
||||
if (isAdded) {
|
||||
this.mcpEmitters.serverEvent.loading.fire(this._getLoadingServerObject(serverName))
|
||||
}
|
||||
return isAdded
|
||||
})
|
||||
const updatedServers = serverNames.filter((serverName) => {
|
||||
const { prevMCPConfig, newMCPConfig } = getPrevAndNewServerConfig(serverName)
|
||||
const isUpdated = prevMCPConfig && newMCPConfig && !equals(prevMCPConfig, newMCPConfig)
|
||||
if (isUpdated) {
|
||||
this.mcpEmitters.serverEvent.loading.fire(this._getLoadingServerObject(serverName))
|
||||
}
|
||||
})
|
||||
const deletedServers = Object.keys(prevServers).filter((serverName) => {
|
||||
const { prevMCPConfig, newMCPConfig } = getPrevAndNewServerConfig(serverName)
|
||||
const isDeleted = prevMCPConfig && !newMCPConfig
|
||||
if (isDeleted) {
|
||||
this.mcpEmitters.serverEvent.loading.fire(this._getLoadingServerObject(serverName))
|
||||
}
|
||||
return isDeleted
|
||||
})
|
||||
|
||||
// Check if no changes were made
|
||||
if (addedServers.length === 0 && updatedServers.length === 0 && deletedServers.length === 0) {
|
||||
console.log('No changes to MCP servers found.')
|
||||
return
|
||||
}
|
||||
|
||||
if (addedServers.length > 0) {
|
||||
// Handle added servers
|
||||
const addPromises: Promise<MCPAddResponse>[] = addedServers.map(async (serverName) => {
|
||||
const addedServer = await this._safeSetupServer(mcpServers[serverName], serverName)
|
||||
return {
|
||||
event: 'add',
|
||||
newServer: addedServer,
|
||||
name: serverName,
|
||||
} as MCPAddResponse
|
||||
});
|
||||
const formattedAddedResponses = await Promise.all(addPromises);
|
||||
formattedAddedResponses.forEach((formattedResponse) => (this.mcpEmitters.serverEvent.add.fire({ response: formattedResponse })));
|
||||
}
|
||||
|
||||
if (updatedServers.length > 0) {
|
||||
// Handle updated servers
|
||||
const updatePromises: Promise<MCPUpdateResponse>[] = updatedServers.map(async (serverName) => {
|
||||
const prevServer = this.clients[serverName]?.formattedServer;
|
||||
const newServer = await this._safeSetupServer(mcpServers[serverName], serverName)
|
||||
return {
|
||||
prevServer,
|
||||
newServer: newServer,
|
||||
event: 'update',
|
||||
name: serverName,
|
||||
} as MCPUpdateResponse
|
||||
});
|
||||
const formattedUpdatedResponses = await Promise.all(updatePromises);
|
||||
formattedUpdatedResponses.forEach((formattedResponse) => (this.mcpEmitters.serverEvent.update.fire({ response: formattedResponse })));
|
||||
}
|
||||
|
||||
if (deletedServers.length > 0) {
|
||||
// Handle deleted servers
|
||||
const deletePromises: Promise<MCPDeleteResponse>[] = deletedServers.map(async (serverName) => {
|
||||
const prevServer = this.clients[serverName]?.formattedServer;
|
||||
await this._callCloseServer(serverName)
|
||||
return {
|
||||
event: 'delete',
|
||||
prevServer,
|
||||
name: serverName,
|
||||
} as MCPDeleteResponse
|
||||
});
|
||||
const formattedDeletedResponses = await Promise.all(deletePromises);
|
||||
formattedDeletedResponses.forEach((formattedResponse) => (this.mcpEmitters.serverEvent.delete.fire({ response: formattedResponse })));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -149,6 +194,7 @@ export class MCPChannel implements IServerChannel {
|
||||
const clientConfig = this.getClientConfig(serverName)
|
||||
const client = new Client(clientConfig)
|
||||
let transport: Transport;
|
||||
let formattedServer: MCPServerModel;
|
||||
|
||||
if (server.url) {
|
||||
// first try HTTP, fall back to SSE
|
||||
@@ -156,11 +202,24 @@ export class MCPChannel implements IServerChannel {
|
||||
transport = new StreamableHTTPClientTransport(server.url);
|
||||
await client.connect(transport);
|
||||
console.log(`Connected via HTTP to ${serverName}`);
|
||||
const { tools } = await client.listTools()
|
||||
formattedServer = {
|
||||
status: 'success',
|
||||
isOn: true,
|
||||
tools: tools,
|
||||
command: server.url.toString(),
|
||||
}
|
||||
} catch (httpErr) {
|
||||
console.warn(`HTTP failed for ${serverName}, trying SSE…`, httpErr);
|
||||
transport = new SSEClientTransport(server.url);
|
||||
await client.connect(transport);
|
||||
console.log(`Connected via SSE to ${serverName}`);
|
||||
formattedServer = {
|
||||
status: 'success',
|
||||
isOn: true,
|
||||
tools: [],
|
||||
command: server.url.toString(),
|
||||
}
|
||||
}
|
||||
} else if (server.command) {
|
||||
console.log('ENV DATA: ', server.env)
|
||||
@@ -173,37 +232,60 @@ export class MCPChannel implements IServerChannel {
|
||||
} as Record<string, string>,
|
||||
});
|
||||
|
||||
client.onerror = (err) => {
|
||||
// TODO: HANDLE SENDING AN EVENT BACK TO THE CLIENT
|
||||
console.error(`Error in MCP client for ${serverName}:`, err);
|
||||
}
|
||||
|
||||
await client.connect(transport)
|
||||
|
||||
console.log(`Connected via Stdio to ${serverName}`);
|
||||
|
||||
// Get the tools from the server
|
||||
const { tools } = await client.listTools()
|
||||
|
||||
// Create a full command string for display
|
||||
const fullCommand = `${server.command} ${server.args?.join(' ') || ''}`
|
||||
console.log('Full Command: ', fullCommand)
|
||||
|
||||
this.mcpEmitters.serverSetup.success.fire({
|
||||
model: {
|
||||
serverName,
|
||||
status: 'success',
|
||||
isOn: true,
|
||||
tools: tools,
|
||||
command: fullCommand,
|
||||
}
|
||||
})
|
||||
// Format server object
|
||||
formattedServer = {
|
||||
status: 'success',
|
||||
isOn: true,
|
||||
tools: tools,
|
||||
command: fullCommand,
|
||||
}
|
||||
|
||||
} else {
|
||||
console.warn(`No url or command for server ${serverName}`);
|
||||
return;
|
||||
throw new Error(`No url or command for server ${serverName}`);
|
||||
}
|
||||
|
||||
// only add to clients map if connect succeeded
|
||||
this.clients[serverName] = { client, mcpConfig: server };
|
||||
|
||||
this.clients[serverName] = { client, mcpConfig: server, formattedServer }
|
||||
return formattedServer;
|
||||
}
|
||||
|
||||
// Helper function to safely setup a server
|
||||
private async _safeSetupServer(serverConfig: MCPServerConfig, serverName: string) {
|
||||
try {
|
||||
return await this._callSetupServer(serverConfig, serverName)
|
||||
} catch (err) {
|
||||
const typedErr = err as Error
|
||||
console.error(`❌ Failed to connect to server "${serverName}":`, err)
|
||||
|
||||
let fullCommand = ''
|
||||
if (serverConfig.command) {
|
||||
fullCommand = `${serverConfig.command} ${serverConfig.args?.join(' ') || ''}`
|
||||
}
|
||||
|
||||
const formattedError: MCPServerErrorModel = {
|
||||
status: 'error',
|
||||
isOn: false,
|
||||
tools: [],
|
||||
error: typedErr.message,
|
||||
command: fullCommand,
|
||||
}
|
||||
|
||||
// Add the error to the clients object
|
||||
this.clients[serverName] = {
|
||||
mcpConfig: serverConfig,
|
||||
formattedServer: formattedError,
|
||||
}
|
||||
|
||||
return formattedError
|
||||
}
|
||||
}
|
||||
|
||||
private async _callCloseAllServers() {
|
||||
@@ -216,28 +298,29 @@ export class MCPChannel implements IServerChannel {
|
||||
private async _callCloseServer(serverName: string) {
|
||||
if (this.clients[serverName]) {
|
||||
const { client } = this.clients[serverName]
|
||||
await client.close()
|
||||
if (client) {
|
||||
await client.close()
|
||||
}
|
||||
delete this.clients[serverName]
|
||||
console.log(`Closed MCP server ${serverName}`);
|
||||
}
|
||||
}
|
||||
|
||||
// Util functions
|
||||
|
||||
// listen functions
|
||||
|
||||
// private _onServerSetupSuccess(serverName: string) {
|
||||
// this.mcpEmitters.serverSetup.success.fire()
|
||||
// }
|
||||
// private _onServerSetupError(error: Error) {
|
||||
// // this.error = error
|
||||
// console.log('WHAAAAT')
|
||||
// console.log('Error in MCPChannel:', error)
|
||||
// }
|
||||
// private _onToolCallSuccess(serverName: string) {
|
||||
// this.mcpEmitters.toolCall.success.fire()
|
||||
// }
|
||||
// private _onToolCallError(serverName: string) {
|
||||
// this.mcpEmitters.toolCall.error.fire()
|
||||
// }
|
||||
private _getLoadingServerObject(serverName: string): MCPServerEventLoadingParam {
|
||||
return {
|
||||
response: {
|
||||
event: 'loading',
|
||||
name: serverName,
|
||||
newServer: {
|
||||
status: 'loading',
|
||||
isOn: false,
|
||||
tools: [],
|
||||
command: '',
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user