diff --git a/sdk/typescript/README.md b/sdk/typescript/README.md index 01085fafa..8dce8ff59 100644 --- a/sdk/typescript/README.md +++ b/sdk/typescript/README.md @@ -7,6 +7,26 @@ The TypeScript SDK provides an idiomatic Node.js interface for building and runn npm install @agentfield/sdk ``` +## Memory event subscriptions + +Pass filters when starting a memory event client to apply them on the server before events are sent over the WebSocket: + +```ts +import { MemoryEventClient } from '@agentfield/sdk'; + +const events = new MemoryEventClient('http://localhost:8080'); +events.onEvent((event) => console.log(event.key)); +events.start({ + patterns: ['user_*', 'session.*'], + scope: 'session', + scopeId: 'session-123' +}); +``` + +`patterns` is a list of memory-key globs and is sent as a comma-separated query parameter. `scope` accepts `workflow`, `session`, `actor`, or `global`, while `scopeId` maps to the server's `scope_id` parameter. All filters are optional, and automatic reconnects reuse the original filters. Omitting them keeps the existing behavior of receiving all events. + +Server-side filtering reduces WebSocket traffic and client-side processing for high-volume event streams. + ## Rate limiting AI calls are wrapped with a stateless rate limiter that matches the Python SDK: exponential backoff, container-scoped jitter, Retry-After support, and a circuit breaker. diff --git a/sdk/typescript/src/memory/MemoryEventClient.ts b/sdk/typescript/src/memory/MemoryEventClient.ts index 6635d60b2..849fbfc80 100644 --- a/sdk/typescript/src/memory/MemoryEventClient.ts +++ b/sdk/typescript/src/memory/MemoryEventClient.ts @@ -4,6 +4,13 @@ import { MemoryClientBase, MemoryRequestOptions } from './MemoryClient.js'; export type MemoryEventHandler = (event: MemoryChangeEvent) => Promise | void; +export interface MemoryEventSubscriptionOptions { + /** Memory key glob patterns to filter on the server. */ + patterns?: string[]; + scope?: MemoryRequestOptions['scope']; + scopeId?: string; +} + export interface MemoryEventHistoryOptions extends MemoryRequestOptions { patterns?: string[], since?: Date, @@ -20,6 +27,7 @@ export class MemoryEventClient extends MemoryClientBase { private reconnectTimer?: ReturnType; private readonly headers: Record; private readonly apiKey?: string; + private subscriptionOptions: MemoryEventSubscriptionOptions = {}; constructor(baseUrl: string, headers?: Record, apiKey?: string) { super(baseUrl, headers); @@ -28,8 +36,13 @@ export class MemoryEventClient extends MemoryClientBase { this.apiKey = apiKey; } - start() { + /** Starts the event stream with optional server-side filters. */ + start(options: MemoryEventSubscriptionOptions = {}) { if (this.ws) return; + this.subscriptionOptions = { + ...options, + patterns: options.patterns ? [...options.patterns] : undefined + }; this.connect(); } @@ -62,7 +75,7 @@ export class MemoryEventClient extends MemoryClientBase { this.cleanup(); this.reconnectPending = false; - this.ws = new WebSocket(this.url, { headers: this.headers }); + this.ws = new WebSocket(this.buildWebSocketUrl(), { headers: this.headers }); this.ws.on('open', () => { this.reconnectDelay = 1000; @@ -99,6 +112,24 @@ export class MemoryEventClient extends MemoryClientBase { }, this.reconnectDelay); } + private buildWebSocketUrl() { + const params = new URLSearchParams(); + const { patterns, scope, scopeId } = this.subscriptionOptions; + + if (patterns && patterns.length > 0) { + params.set('patterns', patterns.join(',')); + } + if (scope) { + params.set('scope', scope); + } + if (scopeId) { + params.set('scope_id', scopeId); + } + + const query = params.toString(); + return query ? `${this.url}?${query}` : this.url; + } + private buildForwardHeaders(headers: Record): Record { const allowed = new Set(['authorization', 'cookie']); const sanitized: Record = {}; diff --git a/sdk/typescript/tests/memory_event_client.test.ts b/sdk/typescript/tests/memory_event_client.test.ts index a4b255f2f..be924b3f3 100644 --- a/sdk/typescript/tests/memory_event_client.test.ts +++ b/sdk/typescript/tests/memory_event_client.test.ts @@ -123,12 +123,32 @@ describe('MemoryEventClient exported methods', () => { }); }); + it('passes subscription filters to the websocket server', () => { + const client = new MemoryEventClient('http://localhost:8080'); + + client.start({ + patterns: ['user_*', 'cart.*'], + scope: 'session', + scopeId: 'session/1 & 2' + }); + + const socketUrl = new URL(MockWebSocket.instances[0].url); + expect(socketUrl.pathname).toBe('/api/v1/memory/events/ws'); + expect(socketUrl.searchParams.get('patterns')).toBe('user_*,cart.*'); + expect(socketUrl.searchParams.get('scope')).toBe('session'); + expect(socketUrl.searchParams.get('scope_id')).toBe('session/1 & 2'); + }); + it('swallows malformed websocket messages and supports reconnect scheduling and stop cleanup', async () => { vi.useFakeTimers(); const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}); const client = new MemoryEventClient('http://localhost:8080'); - client.start(); + client.start({ + patterns: ['memo:*'], + scope: 'workflow', + scopeId: 'workflow-1' + }); const firstSocket = MockWebSocket.instances[0]; await firstSocket.emit('message', Buffer.from('{bad-json')); @@ -143,6 +163,7 @@ describe('MemoryEventClient exported methods', () => { expect(firstSocket.terminate).toHaveBeenCalledTimes(1); const secondSocket = MockWebSocket.instances[1]; + expect(secondSocket.url).toBe(firstSocket.url); client.stop(); expect(secondSocket.terminate).toHaveBeenCalledTimes(1);