Jyh/sse endpoint

demo-server/src/connector.ts

@@ -1,4 +1,10 @@
-import { WebSocketServerEndpoint, ConnectorCenter, WebSocketEndpointServer } from '@opentiny/tiny-agent-mcp-connector';
+import {
+  WebSocketServerEndpoint,
+  ConnectorCenter,
+  WebSocketEndpointServer,
+  SSEEndpointServer,
+  SSEServerEndpoint,
+} from '@opentiny/tiny-agent-mcp-connector';
 import type { Request } from 'express';
 import type { WebSocket } from 'ws';
 
@@ -7,12 +13,31 @@ export function createConnector() {
   const webSocketEndpointServer = new WebSocketEndpointServer({ noServer: true }, connectorCenter);
   webSocketEndpointServer.start();
 
-  const websocketConnectionHandler = async(ws: WebSocket, req: Request) => {
+  const websocketConnectionHandler = async (ws: WebSocket, req: Request) => {
     webSocketEndpointServer.wss.emit('connection', ws, req);
-  }
+  };
   return {
     connectorCenter,
     webSocketEndpointServer,
-    websocketConnectionHandler
+    websocketConnectionHandler,
+  };
+}
+
+export function createSSEConnector() {
+  const connectorCenter = new ConnectorCenter<SSEServerEndpoint>();
+  const port = 8082;
+  const sseEndpointServer = new SSEEndpointServer({ port }, connectorCenter);
+  sseEndpointServer.start();
+
+  try {
+    sseEndpointServer.start();
+  } catch (error) {
+    console.error(`Failed to start SSE endpoint server on port ${port}:`, error);
+    throw error;
   }
+
+  return {
+    connectorCenter,
+    sseEndpointServer,
+  };
 }

demo-server/src/index.ts

@@ -2,7 +2,7 @@ import express, { type Request } from 'express';
 import expressWs from 'express-ws';
 import cors from 'cors';
 import dotenv from 'dotenv';
-import { createConnector } from './connector';
+import { createConnector, createSSEConnector } from './connector';
 import { createProxyServer } from './proxy-server';
 import { createChat } from './chat';
 
@@ -30,22 +30,23 @@ const chatConfigFn = (req: Request) => ({
           'connector-client-id': req.headers['connector-client-id'],
           'mcp-verify-code': req.headers['mcp-verify-code'],
         },
-        timeout: 60
+        timeout: 60,
       },
     },
   },
 });
 
-const { connectorCenter, websocketConnectionHandler } = createConnector();
+// const { connectorCenter, websocketConnectionHandler } = createConnector();
+const { connectorCenter } = createSSEConnector();
 const { sseHandlers, streamableHttpHandlers } = createProxyServer({ connectorCenter });
 const { chatHandler } = createChat(chatConfigFn);
 
 const app = express();
 expressWs(app);
 app.use(cors());
 
-// connector
-(app as unknown as expressWs.WithWebsocketMethod).ws('/ws', websocketConnectionHandler);
+// // connector
+// (app as unknown as expressWs.WithWebsocketMethod).ws('/ws', websocketConnectionHandler);
 
 // mcp server
 app.get('/mcp', streamableHttpHandlers.sessionHandler);

demo/.env.development

@@ -1,2 +1,3 @@
-VITE_CONNECTOR_ENDPOINT_URL=ws://localhost:3001/ws
+VITE_CONNECTOR_WS_ENDPOINT_URL=ws://localhost:3001/ws
+VITE_CONNECTOR_SSE_ENDPOINT_URL=http://localhost:8082/client
 VITE_CHAT_URL=http://localhost:3001/chat

demo/src/mcp/index.js

@@ -1,15 +1,17 @@
-import { EndpointTransport, WebSocketClientEndpoint } from '@opentiny/tiny-agent-mcp-connector';
+import { EndpointTransport, WebSocketClientEndpoint, SSEClientEndpoint } from '@opentiny/tiny-agent-mcp-connector';
 import { McpValidator } from '@opentiny/tiny-agent-mcp-service';
 import { setupMcpService } from '@opentiny/tiny-agent-mcp-service-vue';
 import { McpToolParser } from '@opentiny/tiny-agent-task-mcp';
-import { useTaskScheduler } from "./scheduler";
+import { useTaskScheduler } from './scheduler';
 import mcpToolJson from './mcp-tool.json';
 
 export function initMcp() {
   // Connector
-  const wsEndpoint = new WebSocketClientEndpoint({ url: import.meta.env.VITE_CONNECTOR_ENDPOINT_URL });
-  const endpointTransport = new EndpointTransport(wsEndpoint);
-
+  const sseEndpoint = new SSEClientEndpoint(import.meta.env.VITE_CONNECTOR_SSE_ENDPOINT_URL);
+  const wsEndpoint = new WebSocketClientEndpoint({ url: import.meta.env.VITE_CONNECTOR_WS_ENDPOINT_URL });
+  const endpointTransport = new EndpointTransport(sseEndpoint);
+  // const endpointTransport = new EndpointTransport(wsEndpoint);
+
   // MCP Service
   const mcpService = setupMcpService();
   mcpService.mcpServer.connect(endpointTransport);
@@ -19,22 +21,23 @@ export function initMcp() {
   mcpService.setValidator(mcpValidator);
 
   // Task Scheduler
-  const {taskScheduler, actionManager} = useTaskScheduler();
+  const { taskScheduler, actionManager } = useTaskScheduler();
   const doTask = async (task) => taskScheduler.pushTask(task);
-  
+
   // MCP Tool Parser & mcp-tool.json
   const mcpToolParser = new McpToolParser(doTask);
   mcpToolParser.extractAllTools(mcpToolJson).forEach((tool) => {
     mcpService.mcpServer.registerTool(tool.name, tool.config, tool.cb);
   });
 
   return {
+    sseEndpoint,
     wsEndpoint,
     endpointTransport,
     mcpService,
     mcpValidator,
     taskScheduler,
     actionManager,
-    mcpToolParser
-  }
+    mcpToolParser,
+  };
 }

docs/.vitepress/config.mts

@@ -134,7 +134,7 @@ export default defineConfig({
         {
           text: '连接器扩展',
           base: '/extensions/',
-          items: [{ text: '待补充', link: '' }],
+          items: [{ text: '自定义Connector', link: 'connector' }],
         },
       ],
     },

docs/src/extensions/connector.md

@@ -0,0 +1,245 @@
+# 自定义Connector
+
+本章节主要介绍如何实现一个自定义的通信层Connector，用以客户端与MCP服务端之间进行通信访问
+
+## 自定义服务端
+
+服务端通信需要以 TinyAgent 规定的 IConnectorEndpoint 协议为基础，当然客户端也是基于此协议进行拓展。
+IConnectorEndpoint 协议要求服务端关注消息的接收发送以及启动（初始化）服务端实例，以下是 IConnectorEndpoint 协议的内容
+
+```typescript
+interface IConnectorEndpoint {
+  clientId: string | number;
+  clientIdResolved: Promise<string | number>;
+
+  start(): Promise<void>;
+  close(): Promise<void>;
+
+  send(message: IEndpointMessage): Promise<void>;
+  onmessage?: ((message: IEndpointMessage) => void) | null;
+
+  onclose?: (() => void) | null;
+  onerror?: ((error: Error) => void) | null;
+}
+```
+
+接下来基于 IConnectorEndpoint 协议实现服务端，服务端主要关注于通信，通信方式使用SSE通信方法
+
+```typescript
+import * as http from 'node:http';
+import { JSONRPCMessage } from '@modelcontextprotocol/sdk/types';
+import { EndpointMessageType, IConnectorEndpoint, IEndpointMessage } from '../endpoint.type';
+
+class SSEServerEndpoint implements IConnectorEndpoint {
+  protected app: http.Server;
+  protected res: http.ServerResponse;
+  public clientId: string;
+  public clientIdResolved: Promise<string>;
+
+  constructor(app: http.Server, res: any, clientId: string) {
+    this.app = app;
+    this.res = res;
+    this.clientId = clientId;
+    this.clientIdResolved = Promise.resolve(clientId);
+  }
+
+  async start(): Promise<void> {
+    // 订阅http请求
+    this.app.on('request', (req, res) => {
+      if (req.method === 'OPTIONS' && req.url === '/message') {
+        res.writeHead(204, {
+          'Access-Control-Allow-Origin': '*',
+          'Access-Control-Allow-Methods': '*',
+          'Access-Control-Allow-Headers': '*',
+        });
+        return res.end();
+      }
+      // 定义/message api 用以接收客户端的请求内容
+      if (req.url === '/message') {
+        // 解决跨域问题
+        res.setHeader('Access-Control-Allow-Origin', '*');
+        res.setHeader('Access-Control-Allow-Methods', '*');
+
+        // 读取请求体
+        let body = '';
+        req.on('data', (chunk) => {
+          body += chunk.toString();
+        });
+
+        // 响应请求
+        req.on('end', () => {
+          try {
+            const message = JSON.parse(body);
+            if (message.type === EndpointMessageType.INITIALIZE) {
+              res.end();
+              return;
+            }
+
+            this.onmessage?.(message);
+          } finally {
+            res.end();
+          }
+        });
+      }
+    });
+  }
+  // 关闭SSE连接
+  async close(): Promise<void> {
+    this.res.end();
+  }
+  // 使用SSE连接向客户端推送消息
+  async send(message: IEndpointMessage<JSONRPCMessage>): Promise<void> {
+    this.res.write(`data: ${JSON.stringify(message)}\n\n`);
+  }
+  onmessage?: ((message: IEndpointMessage<JSONRPCMessage>) => void) | null | undefined;
+  onclose?: (() => void) | null | undefined;
+  onerror?: ((error: Error) => void) | null | undefined;
+}
+```
+
+为使模块责任清晰，服务端的启动以及ClientId的初始化工作交给了Server模块，该模块并不负责通信，以下是Server模块的代码示例
+
+```typescript
+import * as http from 'node:http';
+import { ConnectorCenter } from '../connector-center';
+import { EndpointMessageType } from '../endpoint.type';
+import { genId } from '../utils';
+import { SSEServerEndpoint } from './sse-server-endpoint';
+
+class SSEEndpointServer {
+  public app: http.Server;
+  protected port: number;
+  protected connectorCenter: ConnectorCenter<SSEServerEndpoint>;
+
+  constructor(config: { port: number }, connectorCenter: ConnectorCenter<SSEServerEndpoint>) {
+    // 启动http服务器
+    this.app = http.createServer();
+    this.port = config.port;
+    this.connectorCenter = connectorCenter;
+  }
+
+  start() {
+    this.app.listen(config.port);
+    this.app.on('request', (req, res) => {
+      if (req.method === 'OPTIONS' && req.url === '/client') {
+        res.writeHead(204, {
+          'Access-Control-Allow-Origin': '*',
+          'Access-Control-Allow-Methods': '*',
+          'Access-Control-Allow-Headers': '*',
+        });
+        return res.end();
+      }
+      // 定义/client api用以初始化ClientId
+      if (req.url === '/client') {
+        const clientId = genId();
+        const endpoint = new SSEServerEndpoint(this.app, res, clientId);
+
+        endpoint.start();
+
+        this.connectorCenter.setClient(clientId, endpoint);
+
+        // SSE方式通信
+        res.setHeader('Content-Type', 'text/event-stream');
+        res.setHeader('Cache-Control', 'no-cache');
+        res.setHeader('Connection', 'keep-alive');
+        // 解决跨域问题
+        res.setHeader('Access-Control-Allow-Origin', '*');
+        res.setHeader('Access-Control-Allow-Methods', '*');
+        // 推送初始化的ClientId
+        res.write(
+          'data: ' +
+            JSON.stringify({
+              type: EndpointMessageType.INITIALIZE,
+              data: {
+                clientId,
+              },
+            }) +
+            '\n\n',
+        );
+      }
+    });
+  }
+}
+```
+
+## 自定义客户端
+
+下面是SSE客户端的实现，同意基于IConnectorEndpoint 协议
+
+```typescript
+import { JSONRPCMessage } from '@modelcontextprotocol/sdk/types';
+import { EndpointMessageType, IConnectorEndpoint, IEndpointMessage } from '../endpoint.type';
+
+class SSEClientEndpoint implements IConnectorEndpoint {
+  public clientId!: string | number;
+  public clientIdResolved: Promise<string | number>;
+  protected clientIdResolver!: (id: string | number) => void;
+  // SSE连接实例
+  protected eventSource!: EventSource;
+  // SSE连接url
+  protected url: string;
+  // SSE连接配置项
+  protected config?: EventSourceInit;
+
+  constructor(url: string, config?: EventSourceInit) {
+    this.url = url;
+    this.config = config;
+    this.clientIdResolved = new Promise((resolve) => {
+      this.clientIdResolver = resolve;
+    });
+  }
+  start(): Promise<void> {
+    return new Promise((resolve) => {
+      // 创建SSE连接实例
+      this.eventSource = new EventSource(this.url, this.config);
+
+      this.eventSource.onopen = () => {};
+
+      this.eventSource.onerror = (error) => {
+        console.error('SSE error:', error);
+        this.onerror?.(error as any);
+      };
+
+      // 接收服务端消息
+      this.eventSource.onmessage = (messageEvent: MessageEvent<string>) => {
+        const message: IEndpointMessage = JSON.parse(messageEvent.data);
+        if (message.type === EndpointMessageType.INITIALIZE) {
+          this.clientId = (message.data as any).clientId;
+          this.clientIdResolver(this.clientId);
+          resolve();
+          return;
+        }
+        this.onmessage?.(message);
+      };
+    });
+  }
+  async close(): Promise<void> {
+    this.eventSource.close();
+    this.onclose?.();
+  }
+  // 由于SSE是用以服务端向客户端推送消息，后续客户端主动请求则是以HTTP进行
+  async send(message: IEndpointMessage<JSONRPCMessage>): Promise<void> {
+    try {
+      if (message.type !== EndpointMessageType.INITIALIZE) {
+        await this.clientIdResolved;
+      }
+
+      const url = URL.parse(this.url);
+
+      // 请求/message api 向服务端发送请求
+      fetch(`${url?.origin}/message`, {
+        method: 'POST',
+        body: JSON.stringify(message),
+        headers: {
+          'Content-Type': 'application/json',
+        },
+      });
+    } catch (e) {
+      this.onerror?.(e as Error);
+    }
+  }
+  onmessage?: ((message: IEndpointMessage<JSONRPCMessage>) => void) | null | undefined;
+  onclose?: (() => void) | null | undefined;
+  onerror?: ((error: Error) => void) | null | undefined;
+}
+```