diff --git a/README.md b/README.md
index b29d2ae..7003e76 100644
--- a/README.md
+++ b/README.md
@@ -19,6 +19,8 @@ npm install @sinclair/typebox
 ## Features
 
 - **Complete MCP 2025-06-18 Support**: Implements the full Model Context Protocol specification with elicitation
+- **Async Iterator Streaming**: Real-time streaming of tool results via Server-Sent Events
+- **Per-Stream Event IDs**: MCP specification compliant event ID architecture for proper resumability
 - **Elicitation Support**: Server-to-client information requests with schema validation
 - **TypeBox Validation**: Type-safe schema validation with automatic TypeScript inference
 - **Security Enhancements**: Input sanitization, rate limiting, and security assessment
@@ -122,6 +124,179 @@ app.mcpAddPrompt({
 await app.listen({ port: 3000 })
 ```
 
+## Async Iterator Streaming
+
+The plugin supports **real-time streaming** of tool results via Server-Sent Events (SSE). Tools that return async iterators automatically stream their results as separate SSE events, enabling real-time progress updates and streaming data processing.
+
+### Basic Streaming Tool
+
+```typescript
+import Fastify from 'fastify'
+import mcpPlugin from '@platformatic/mcp'
+
+const app = Fastify({ logger: true })
+
+await app.register(mcpPlugin, {
+  enableSSE: true, // Required for streaming
+  serverInfo: { name: 'streaming-server', version: '1.0.0' },
+  capabilities: { tools: {} }
+})
+
+// Regular tool (returns JSON immediately)
+app.mcpAddTool({
+  name: 'immediate_response',
+  description: 'Returns immediate result',
+  inputSchema: {
+    type: 'object',
+    properties: {
+      message: { type: 'string' }
+    }
+  }
+}, async (params) => {
+  return {
+    content: [{ type: 'text', text: `Echo: ${params.message}` }]
+  }
+})
+
+// Streaming tool (returns SSE stream)
+app.mcpAddTool({
+  name: 'streaming_counter',
+  description: 'Streams counting progress',
+  inputSchema: {
+    type: 'object',
+    properties: {
+      count: { type: 'number', minimum: 1, maximum: 10 }
+    },
+    required: ['count']
+  }
+}, async function* (params) {
+  // Yield incremental progress updates
+  for (let i = 1; i <= params.count; i++) {
+    yield {
+      content: [{
+        type: 'text',
+        text: `Processing step ${i}/${params.count}...`
+      }]
+    }
+    
+    // Simulate async work
+    await new Promise(resolve => setTimeout(resolve, 500))
+  }
+  
+  // Final result
+  return {
+    content: [{
+      type: 'text',
+      text: `✅ Completed all ${params.count} steps!`
+    }]
+  }
+})
+
+await app.listen({ port: 3000 })
+```
+
+### Streaming Response Format
+
+When a tool returns an async iterator, the plugin automatically:
+
+1. **Detects the async iterator** using `Symbol.asyncIterator`
+2. **Changes response type** to `Content-Type: text/event-stream`
+3. **Streams each yielded value** as a separate SSE event
+4. **Sends the final return value** as the last event
+5. **Handles errors gracefully** during streaming
+
+### Client Usage
+
+```bash
+# Regular tool (returns JSON)
+curl -X POST http://localhost:3000/mcp \
+  -H "Content-Type: application/json" \
+  -d '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"immediate_response","arguments":{"message":"Hello"}}}'
+
+# Streaming tool (returns text/event-stream)
+curl -X POST http://localhost:3000/mcp \
+  -H "Content-Type: application/json" \
+  -d '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"streaming_counter","arguments":{"count":3}}}'
+```
+
+### Advanced Streaming Examples
+
+#### File Processing Stream
+```typescript
+app.mcpAddTool({
+  name: 'process_files',
+  description: 'Process multiple files with progress updates'
+}, async function* (params) {
+  for (const [index, filename] of params.files.entries()) {
+    yield {
+      content: [{
+        type: 'text',
+        text: `📁 Processing file ${index + 1}/${params.files.length}: ${filename}`
+      }]
+    }
+    
+    // Simulate file processing
+    await processFile(filename)
+    
+    yield {
+      content: [{
+        type: 'text',
+        text: `✅ Completed: ${filename}`
+      }]
+    }
+  }
+  
+  return {
+    content: [{
+      type: 'text',
+      text: `🎉 All ${params.files.length} files processed!`
+    }]
+  }
+})
+```
+
+#### Error Handling During Streaming
+```typescript
+app.mcpAddTool({
+  name: 'streaming_with_errors',
+  description: 'Demonstrates error handling in streams'
+}, async function* (params) {
+  try {
+    for (let i = 1; i <= 5; i++) {
+      if (i === 3) {
+        throw new Error('Simulated processing error')
+      }
+      
+      yield {
+        content: [{
+          type: 'text',
+          text: `Step ${i}: Working...`
+        }]
+      }
+      
+      await new Promise(resolve => setTimeout(resolve, 300))
+    }
+    
+    return {
+      content: [{ type: 'text', text: 'All steps completed' }]
+    }
+  } catch (error) {
+    // Errors during streaming are handled gracefully
+    // Client receives all yielded values before the error
+    throw error
+  }
+})
+```
+
+### Key Features
+
+- **🔄 Automatic Detection**: No configuration needed - just return an async generator
+- **📡 Real-time Updates**: Each `yield` becomes an immediate SSE event
+- **🛡️ Error Handling**: Partial results preserved, errors handled gracefully
+- **🔙 Backward Compatible**: Existing tools continue to work unchanged
+- **⚡ Performance**: Efficient streaming with proper event ID management
+- **🌊 MCP Compliant**: Per-stream event IDs for proper resumability
+
 ## Elicitation Support (MCP 2025-06-18)
 
 The plugin supports the elicitation capability, allowing servers to request structured information from clients. This enables dynamic data collection with schema validation.
diff --git a/examples/streaming-demo.ts b/examples/streaming-demo.ts
new file mode 100644
index 0000000..eb4dfc2
--- /dev/null
+++ b/examples/streaming-demo.ts
@@ -0,0 +1,187 @@
+#!/usr/bin/env node
+
+import Fastify from 'fastify'
+import mcpPlugin from '../dist/index.js'
+
+const app = Fastify({ logger: true })
+
+// Register MCP plugin with SSE enabled for streaming support
+await app.register(mcpPlugin, {
+  serverInfo: { name: 'streaming-demo', version: '1.0.0' },
+  enableSSE: true
+})
+
+// Regular tool that returns immediate results
+app.mcpAddTool({
+  name: 'immediate_response',
+  description: 'Tool that returns an immediate response',
+  inputSchema: {
+    type: 'object',
+    properties: {
+      message: { type: 'string' }
+    },
+    required: ['message']
+  }
+}, async (params) => {
+  return {
+    content: [{ type: 'text', text: `Immediate: ${params.message}` }]
+  }
+})
+
+// Streaming tool using async generator
+app.mcpAddTool({
+  name: 'streaming_response',
+  description: 'Tool that streams responses using async generator',
+  inputSchema: {
+    type: 'object',
+    properties: {
+      count: { type: 'number', minimum: 1, maximum: 10 },
+      delay: { type: 'number', minimum: 100, maximum: 2000, default: 500 }
+    },
+    required: ['count']
+  }
+}, async function * (params) {
+  const delay = params.delay ?? 500
+
+  // Yield incremental chunks
+  for (let i = 1; i <= params.count; i++) {
+    yield {
+      content: [{
+        type: 'text',
+        text: `Streaming chunk ${i}/${params.count}: Processing...`
+      }]
+    }
+
+    // Simulate async work
+    await new Promise(resolve => setTimeout(resolve, delay))
+  }
+
+  // Final result
+  return {
+    content: [{
+      type: 'text',
+      text: `✅ Completed all ${params.count} processing steps!`
+    }]
+  }
+})
+
+// Streaming tool that simulates file processing
+app.mcpAddTool({
+  name: 'file_processor',
+  description: 'Simulates processing multiple files with streaming updates',
+  inputSchema: {
+    type: 'object',
+    properties: {
+      files: {
+        type: 'array',
+        items: { type: 'string' },
+        minItems: 1,
+        maxItems: 5
+      }
+    },
+    required: ['files']
+  }
+}, async function * (params) {
+  for (const [index, filename] of params.files.entries()) {
+    // Simulate processing each file
+    yield {
+      content: [{
+        type: 'text',
+        text: `📁 Processing file ${index + 1}/${params.files.length}: ${filename}`
+      }]
+    }
+
+    // Simulate processing time
+    await new Promise(resolve => setTimeout(resolve, 800))
+
+    yield {
+      content: [{
+        type: 'text',
+        text: `✅ Completed processing: ${filename}`
+      }]
+    }
+  }
+
+  // Final summary
+  return {
+    content: [{
+      type: 'text',
+      text: `🎉 All ${params.files.length} files processed successfully!`
+    }]
+  }
+})
+
+// Error demonstration tool
+app.mcpAddTool({
+  name: 'error_demo',
+  description: 'Demonstrates error handling in streaming',
+  inputSchema: {
+    type: 'object',
+    properties: {
+      errorAfter: { type: 'number', minimum: 1, maximum: 5, default: 3 }
+    }
+  }
+}, async function * (params) {
+  const errorAfter = params.errorAfter ?? 3
+
+  for (let i = 1; i <= 5; i++) {
+    if (i === errorAfter) {
+      throw new Error(`Simulated error at step ${i}`)
+    }
+
+    yield {
+      content: [{
+        type: 'text',
+        text: `Step ${i}: Everything working fine...`
+      }]
+    }
+
+    await new Promise(resolve => setTimeout(resolve, 300))
+  }
+
+  return {
+    content: [{
+      type: 'text',
+      text: 'This should not be reached due to the error'
+    }]
+  }
+})
+
+// Start the server
+const port = parseInt(process.env.PORT || '3000', 10)
+const host = process.env.HOST || '127.0.0.1'
+
+try {
+  await app.listen({ port, host })
+  console.log(`🚀 MCP Streaming Demo Server running on http://${host}:${port}`)
+  console.log('\n📖 Usage Examples:')
+  console.log(`
+  # Test immediate response (returns JSON)
+  curl -X POST http://${host}:${port}/mcp \\
+    -H "Content-Type: application/json" \\
+    -d '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"immediate_response","arguments":{"message":"Hello World"}}}'
+
+  # Test streaming response (returns text/event-stream)
+  curl -X POST http://${host}:${port}/mcp \\
+    -H "Content-Type: application/json" \\
+    -d '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"streaming_response","arguments":{"count":3,"delay":1000}}}'
+
+  # Test file processing simulation
+  curl -X POST http://${host}:${port}/mcp \\
+    -H "Content-Type: application/json" \\
+    -d '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"file_processor","arguments":{"files":["doc1.pdf","image.jpg","data.csv"]}}}'
+
+  # Test error handling
+  curl -X POST http://${host}:${port}/mcp \\
+    -H "Content-Type: application/json" \\
+    -d '{"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"error_demo","arguments":{"errorAfter":2}}}'
+
+  # List all available tools
+  curl -X POST http://${host}:${port}/mcp \\
+    -H "Content-Type: application/json" \\
+    -d '{"jsonrpc":"2.0","id":5,"method":"tools/list","params":{}}'
+  `)
+} catch (err) {
+  app.log.error(err)
+  process.exit(1)
+}
diff --git a/src/handlers.ts b/src/handlers.ts
index bc4d33d..0b2134b 100644
--- a/src/handlers.ts
+++ b/src/handlers.ts
@@ -57,6 +57,21 @@ export function createError (id: string | number, code: number, message: string,
   }
 }
 
+// Helper function to check if a value is an async iterator
+function isAsyncIterator (value: any): value is AsyncGenerator<CallToolResult, CallToolResult, unknown> {
+  return value != null &&
+         typeof value === 'object' &&
+         typeof value[Symbol.asyncIterator] === 'function' &&
+         typeof value.next === 'function'
+}
+
+// Interface for streaming responses to differentiate from regular responses
+export interface StreamingToolResponse {
+  isStreaming: true
+  iterator: AsyncGenerator<CallToolResult, CallToolResult, unknown>
+  requestId: string | number
+}
+
 function handleInitialize (request: JSONRPCRequest, dependencies: HandlerDependencies): JSONRPCResponse {
   const { opts, capabilities, serverInfo } = dependencies
   const result: InitializeResult = {
@@ -114,7 +129,7 @@ async function handleToolsCall (
   request: JSONRPCRequest,
   sessionId: string | undefined,
   dependencies: HandlerDependencies
-): Promise<JSONRPCResponse | JSONRPCError> {
+): Promise<JSONRPCResponse | JSONRPCError | StreamingToolResponse> {
   const { tools } = dependencies
 
   // Validate the request parameters structure
@@ -191,6 +206,16 @@ async function handleToolsCall (
       // Use validated arguments
       try {
         const result = await tool.handler(argumentsValidation.data, { sessionId, request: dependencies.request, reply: dependencies.reply, authContext: dependencies.authContext })
+
+        // Check if result is an async iterator for streaming
+        if (isAsyncIterator(result)) {
+          return {
+            isStreaming: true,
+            iterator: result,
+            requestId: request.id
+          }
+        }
+
         return createResponse(request.id, result)
       } catch (error: any) {
         const result: CallToolResult = {
@@ -206,6 +231,16 @@ async function handleToolsCall (
       // Regular JSON Schema - basic validation or pass through
       try {
         const result = await tool.handler(toolArguments, { sessionId, request: dependencies.request, reply: dependencies.reply, authContext: dependencies.authContext })
+
+        // Check if result is an async iterator for streaming
+        if (isAsyncIterator(result)) {
+          return {
+            isStreaming: true,
+            iterator: result,
+            requestId: request.id
+          }
+        }
+
         return createResponse(request.id, result)
       } catch (error: any) {
         const result: CallToolResult = {
@@ -227,6 +262,16 @@ async function handleToolsCall (
         reply: dependencies.reply,
         authContext: dependencies.authContext
       })
+
+      // Check if result is an async iterator for streaming
+      if (isAsyncIterator(result)) {
+        return {
+          isStreaming: true,
+          iterator: result,
+          requestId: request.id
+        }
+      }
+
       return createResponse(request.id, result)
     } catch (error: any) {
       const result: CallToolResult = {
@@ -444,7 +489,7 @@ export async function handleRequest (
   request: JSONRPCRequest,
   sessionId: string | undefined,
   dependencies: HandlerDependencies
-): Promise<JSONRPCResponse | JSONRPCError> {
+): Promise<JSONRPCResponse | JSONRPCError | StreamingToolResponse> {
   const { app } = dependencies
 
   app.log.info({
@@ -496,7 +541,7 @@ export async function processMessage (
   message: JSONRPCMessage,
   sessionId: string | undefined,
   dependencies: HandlerDependencies
-): Promise<JSONRPCResponse | JSONRPCError | null> {
+): Promise<JSONRPCResponse | JSONRPCError | StreamingToolResponse | null> {
   if ('id' in message && 'method' in message) {
     return await handleRequest(message as JSONRPCRequest, sessionId, dependencies)
   } else if ('method' in message) {
diff --git a/src/routes/auth-routes.ts b/src/routes/auth-routes.ts
index 2180c0d..6277aa4 100644
--- a/src/routes/auth-routes.ts
+++ b/src/routes/auth-routes.ts
@@ -107,10 +107,10 @@ const authRoutesPlugin: FastifyPluginAsync<AuthRoutesOptions> = async (fastify:
       // Create session metadata with auth session data
       const sessionMetadata = {
         id: authRequest.state,
-        eventId: 0,
         createdAt: new Date(),
         lastActivity: new Date(),
-        authSession: sessionData
+        authSession: sessionData,
+        streams: new Map()
       }
 
       await sessionStore.create(sessionMetadata)
diff --git a/src/routes/mcp.ts b/src/routes/mcp.ts
index 37fe54e..5885324 100644
--- a/src/routes/mcp.ts
+++ b/src/routes/mcp.ts
@@ -7,7 +7,7 @@ import type { MCPPluginOptions, MCPTool, MCPResource, MCPPrompt } from '../types
 import type { SessionStore, SessionMetadata } from '../stores/session-store.ts'
 import type { MessageBroker } from '../brokers/message-broker.ts'
 import type { AuthorizationContext } from '../types/auth-types.ts'
-import { processMessage } from '../handlers.ts'
+import { processMessage, createResponse, createError, type StreamingToolResponse } from '../handlers.ts'
 
 interface MCPPubSubRoutesOptions {
   enableSSE: boolean
@@ -22,6 +22,11 @@ interface MCPPubSubRoutesOptions {
   localStreams: Map<string, Set<any>>
 }
 
+// Helper function to check if response is streaming
+function isStreamingResponse (response: any): response is StreamingToolResponse {
+  return response && response.isStreaming === true && response.iterator && response.requestId !== undefined
+}
+
 const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async (app, options) => {
   const { enableSSE, opts, capabilities, serverInfo, tools, resources, prompts, sessionStore, messageBroker, localStreams } = options
 
@@ -29,10 +34,9 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async
     const sessionId = randomUUID()
     const session: SessionMetadata = {
       id: sessionId,
-      eventId: 0,
-      lastEventId: undefined,
       createdAt: new Date(),
-      lastActivity: new Date()
+      lastActivity: new Date(),
+      streams: new Map()
     }
 
     await sessionStore.create(session)
@@ -43,17 +47,12 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async
       const streams = localStreams.get(sessionId)
       if (streams && streams.size > 0) {
         app.log.debug({ sessionId, message }, 'Received message for session via broker, sending to streams')
-        sendSSEToStreams(sessionId, message, streams)
+        await sendSSEToStreams(sessionId, message, streams)
       } else {
         app.log.debug({ sessionId }, 'Received message for session via broker, storing in history without active streams')
-        // Store message in history even without active streams for session persistence
-        const session = await sessionStore.get(sessionId)
-        if (session) {
-          const eventId = (++session.eventId).toString()
-          session.lastEventId = eventId
-          session.lastActivity = new Date()
-          await sessionStore.addMessage(sessionId, eventId, message)
-        }
+        // For backward compatibility, store in session-level history if no streams are active
+        // This maintains existing behavior for legacy usage
+        await sessionStore.addSessionMessage(sessionId, '0', message)
       }
     })
 
@@ -65,35 +64,100 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async
     return accept ? accept.includes('text/event-stream') : false
   }
 
-  function hasActiveSSESession (sessionId?: string): boolean {
-    if (!sessionId) return false
-    const streams = localStreams.get(sessionId)
-    return streams ? streams.size > 0 : false
-  }
-
   async function sendSSEToStreams (sessionId: string, message: JSONRPCMessage, streams: Set<FastifyReply>): Promise<void> {
-    const session = await sessionStore.get(sessionId)
-    if (!session) return
+    // Check if this is a broadcast notification or elicitation (these should use session-level storage)
+    const isBroadcast = 'method' in message && (
+      message.method === 'notifications/message' ||
+      message.method.startsWith('notifications/') ||
+      message.method === 'elicitation/create'
+    )
+
+    if (isBroadcast) {
+      // Use broadcast method for broadcast notifications and elicitation requests
+      await sendSSEToStreamsBroadcast(sessionId, message, streams)
+      return
+    }
 
-    const eventId = (++session.eventId).toString()
-    const sseEvent = `id: ${eventId}\ndata: ${JSON.stringify(message)}\n\n`
-    session.lastEventId = eventId
-    session.lastActivity = new Date()
+    // According to MCP spec line 145: server MUST send each message to only one stream
+    // For now, we'll select the first available stream (round-robin could be implemented later)
+    const streamArray = Array.from(streams)
+    if (streamArray.length === 0) return
 
-    // Store message in history
-    await sessionStore.addMessage(sessionId, eventId, message)
+    // Select the first stream for this message (simple strategy)
+    const selectedStream = streamArray[0]
+    const streamId = (selectedStream as any).mcpStreamId
+
+    if (!streamId) {
+      app.log.warn('Stream missing mcpStreamId, falling back to broadcast')
+      // Fallback to broadcast behavior if streamId is missing
+      await sendSSEToStreamsBroadcast(sessionId, message, streams)
+      return
+    }
+
+    try {
+      // Get current stream metadata to determine next event ID
+      const streamMetadata = await sessionStore.getStream(sessionId, streamId)
+      if (!streamMetadata) {
+        app.log.warn(`Stream metadata not found for stream: ${streamId}`)
+        return
+      }
+
+      // Generate next event ID for this specific stream
+      const eventId = (streamMetadata.eventId + 1).toString()
+      const sseEvent = `id: ${eventId}\ndata: ${JSON.stringify(message)}\n\n`
+
+      // Send to the selected stream
+      selectedStream.raw.write(sseEvent)
+
+      // Store message in per-stream history
+      await sessionStore.addMessage(sessionId, streamId, eventId, message)
+      await sessionStore.updateStreamActivity(sessionId, streamId)
+
+      app.log.debug({
+        sessionId,
+        streamId,
+        eventId,
+        messageType: 'method' in message ? message.method : 'response'
+      }, 'Sent message to specific stream')
+    } catch (error) {
+      app.log.error({ err: error, sessionId, streamId }, 'Failed to send SSE event to stream')
 
-    // Send to all connected streams in this session
+      // Remove dead stream
+      streams.delete(selectedStream)
+
+      // Clean up session if no streams left
+      if (streams.size === 0) {
+        app.log.info({ sessionId }, 'Session has no active streams, cleaning up')
+        localStreams.delete(sessionId)
+        await messageBroker.unsubscribe(`mcp/session/${sessionId}/message`)
+      }
+    }
+  }
+
+  async function sendSSEToStreamsBroadcast (sessionId: string, message: JSONRPCMessage, streams: Set<FastifyReply>): Promise<void> {
+    // Broadcast method for notifications and elicitation - stores in session-level history
     const deadStreams = new Set<FastifyReply>()
+
+    // Use timestamp-based event ID for broadcast compatibility
+    const eventId = Date.now().toString()
+    const sseEvent = `id: ${eventId}\ndata: ${JSON.stringify(message)}\n\n`
+
     for (const stream of streams) {
       try {
         stream.raw.write(sseEvent)
       } catch (error) {
-        app.log.error({ err: error }, 'Failed to write SSE event')
+        app.log.error({ err: error }, 'Failed to write legacy SSE event')
         deadStreams.add(stream)
       }
     }
 
+    // Store message in session-level history (broadcast messages are session-wide)
+    try {
+      await sessionStore.addSessionMessage(sessionId, eventId, message)
+    } catch (error) {
+      app.log.error({ err: error }, 'Failed to store broadcast session message')
+    }
+
     // Clean up dead streams
     for (const deadStream of deadStreams) {
       streams.delete(deadStream)
@@ -101,33 +165,118 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async
 
     // Clean up session if no streams left
     if (streams.size === 0) {
-      app.log.info({
-        sessionId
-      }, 'Session has no active streams, cleaning up')
+      app.log.info({ sessionId }, 'Session has no active streams, cleaning up')
       localStreams.delete(sessionId)
       await messageBroker.unsubscribe(`mcp/session/${sessionId}/message`)
     }
   }
 
-  async function replayMessagesFromEventId (sessionId: string, lastEventId: string, stream: FastifyReply): Promise<void> {
+  async function handleStreamingResponse (
+    streamingResponse: StreamingToolResponse,
+    sessionId: string | undefined,
+    reply: FastifyReply
+  ): Promise<void> {
+    // Hijack the response for streaming
+    reply.hijack()
+    const raw = reply.raw
+
+    // Set SSE headers
+    raw.setHeader('Content-Type', 'text/event-stream')
+    raw.setHeader('Cache-Control', 'no-cache')
+    raw.writeHead(200)
+
+    let eventId = 1
+
     try {
-      const messagesToReplay = await sessionStore.getMessagesFrom(sessionId, lastEventId)
+      // Manually iterate through async iterator to capture both yielded values and return value
+      const iterator = streamingResponse.iterator
+      let result = await iterator.next()
+
+      while (!result.done) {
+        // Handle yielded values
+        const response = createResponse(streamingResponse.requestId, result.value)
+        const sseEvent = `id: ${eventId}\ndata: ${JSON.stringify(response)}\n\n`
+
+        try {
+          raw.write(sseEvent)
+        } catch (error) {
+          app.log.error({ err: error }, 'Failed to write SSE chunk')
+          break
+        }
+
+        // Update session if available - use legacy method for backward compatibility
+        if (enableSSE && sessionId) {
+          await sessionStore.addSessionMessage(sessionId, eventId.toString(), response)
+        }
+
+        eventId++
+        result = await iterator.next()
+      }
+
+      // Handle final return value if present
+      if (result.value !== undefined) {
+        const response = createResponse(streamingResponse.requestId, result.value)
+        const sseEvent = `id: ${eventId}\ndata: ${JSON.stringify(response)}\n\n`
+
+        try {
+          raw.write(sseEvent)
+        } catch (error) {
+          app.log.error({ err: error }, 'Failed to write final SSE event')
+        }
+
+        // Update session with final value if available - use legacy method for backward compatibility
+        if (enableSSE && sessionId) {
+          await sessionStore.addSessionMessage(sessionId, eventId.toString(), response)
+        }
+      }
+    } catch (error: any) {
+      // Send error event
+      const errorResponse = createError(
+        streamingResponse.requestId,
+        INTERNAL_ERROR,
+        `Streaming error: ${error.message || error}`
+      )
+      const errorEvent = `id: ${eventId}\ndata: ${JSON.stringify(errorResponse)}\n\n`
+
+      try {
+        raw.write(errorEvent)
+      } catch (writeError) {
+        app.log.error({ err: writeError }, 'Failed to write error event')
+      }
+
+      // Update session with error if available - use legacy method for backward compatibility
+      if (enableSSE && sessionId) {
+        await sessionStore.addSessionMessage(sessionId, eventId.toString(), errorResponse)
+      }
+    } finally {
+      // Close the stream
+      try {
+        raw.end()
+      } catch (error) {
+        app.log.error({ err: error }, 'Failed to close SSE stream')
+      }
+    }
+  }
+
+  async function replayStreamMessagesFromEventId (sessionId: string, streamId: string, lastEventId: string, stream: FastifyReply): Promise<void> {
+    try {
+      const messagesToReplay = await sessionStore.getMessagesFrom(sessionId, streamId, lastEventId)
 
       for (const entry of messagesToReplay) {
         const sseEvent = `id: ${entry.eventId}\ndata: ${JSON.stringify(entry.message)}\n\n`
         try {
           stream.raw.write(sseEvent)
         } catch (error) {
-          app.log.error({ err: error }, 'Failed to replay SSE event')
+          app.log.error({ err: error }, 'Failed to replay per-stream SSE event')
           break
         }
       }
 
       if (messagesToReplay.length > 0) {
-        app.log.info(`Replayed ${messagesToReplay.length} messages from event ID: ${lastEventId}`)
+        app.log.info(`Replayed ${messagesToReplay.length} messages from event ID: ${lastEventId} for stream: ${streamId}`)
       }
     } catch (error) {
-      app.log.warn({ err: error, lastEventId }, 'Failed to replay messages from event ID')
+      app.log.warn({ err: error, lastEventId, streamId }, 'Failed to replay per-stream messages from event ID')
     }
   }
 
@@ -190,6 +339,13 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async
         authContext
       })
       if (response) {
+        // Check if this is a streaming response
+        if (isStreamingResponse(response)) {
+          // Handle streaming response
+          await handleStreamingResponse(response, sessionId, reply)
+          return // Response already sent via streaming
+        }
+
         return response
       } else {
         reply.code(202)
@@ -223,13 +379,8 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async
       const sessionId = (request.headers['mcp-session-id'] as string) ||
                        (request.query as any)['mcp-session-id']
 
-      // Check if there's already an active SSE session
-      if (hasActiveSSESession(sessionId)) {
-        reply.type('application/json').code(409).send({
-          error: 'Conflict: SSE session already active for this session ID'
-        })
-        return
-      }
+      // Note: According to MCP spec line 143, clients MAY remain connected to multiple SSE streams simultaneously
+      // So we allow multiple streams per session
 
       request.log.info({ sessionId }, 'Handling SSE request')
 
@@ -238,9 +389,7 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async
 
       const raw = reply.raw
 
-      // Set up SSE stream
-      raw.setHeader('Content-type', 'text/event-stream')
-      raw.setHeader('Cache-Control', 'no-cache')
+      // Headers will be set later with stream ID
 
       let session: SessionMetadata
       if (sessionId) {
@@ -256,6 +405,21 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async
         raw.setHeader('Mcp-Session-Id', session.id)
       }
 
+      // Generate unique stream ID for this SSE connection
+      const streamId = randomUUID()
+
+      // Create stream metadata for per-stream event ID tracking
+      const streamMetadata = await sessionStore.createStream(session.id, streamId)
+      if (!streamMetadata) {
+        raw.writeHead(500)
+        raw.end('Failed to create stream')
+        return
+      }
+
+      // Set headers before writing head
+      raw.setHeader('Content-type', 'text/event-stream')
+      raw.setHeader('Cache-Control', 'no-cache')
+      raw.setHeader('Mcp-Stream-Id', streamId)
       raw.writeHead(200)
 
       let streams = localStreams.get(session.id)
@@ -265,17 +429,21 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async
       }
       streams.add(reply)
 
+      // Associate the reply with the stream ID for per-stream management
+      ;(reply as any).mcpStreamId = streamId
+
       app.log.info({
         sessionId: session.id,
+        streamId,
         totalStreams: streams.size,
         method: 'GET'
       }, 'Added new stream to session')
 
-      // Handle resumability with Last-Event-ID
+      // Handle resumability with Last-Event-ID - now per-stream
       const lastEventId = request.headers['last-event-id'] as string
       if (lastEventId) {
-        app.log.info(`Resuming SSE stream from event ID: ${lastEventId}`)
-        await replayMessagesFromEventId(session.id, lastEventId, reply)
+        app.log.info(`Resuming SSE stream from event ID: ${lastEventId} for stream: ${streamId}`)
+        await replayStreamMessagesFromEventId(session.id, streamId, lastEventId, reply)
       }
 
       // Handle connection close
@@ -285,9 +453,13 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async
           streams.delete(reply)
           app.log.info({
             sessionId: session.id,
+            streamId,
             remainingStreams: streams.size
           }, 'SSE connection closed')
 
+          // Clean up stream metadata
+          sessionStore.deleteStream(session.id, streamId)
+
           if (streams.size === 0) {
             app.log.info({
               sessionId: session.id
@@ -317,7 +489,8 @@ const mcpPubSubRoutesPlugin: FastifyPluginAsync<MCPPubSubRoutesOptions> = async
 
       reply.raw.on('close', () => {
         app.log.info({
-          sessionId: session.id
+          sessionId: session.id,
+          streamId
         }, 'SSE heartbeat connection closed')
         clearInterval(heartbeatInterval)
       })
diff --git a/src/stores/memory-session-store.ts b/src/stores/memory-session-store.ts
index 9dbeb4c..ec9f133 100644
--- a/src/stores/memory-session-store.ts
+++ b/src/stores/memory-session-store.ts
@@ -1,5 +1,5 @@
 import type { JSONRPCMessage } from '../schema.ts'
-import type { SessionStore, SessionMetadata } from './session-store.ts'
+import type { SessionStore, SessionMetadata, StreamMetadata } from './session-store.ts'
 import type { AuthorizationContext, TokenRefreshInfo } from '../types/auth-types.ts'
 
 interface MessageHistoryEntry {
@@ -9,7 +9,8 @@ interface MessageHistoryEntry {
 
 export class MemorySessionStore implements SessionStore {
   private sessions = new Map<string, SessionMetadata>()
-  private messageHistory = new Map<string, MessageHistoryEntry[]>()
+  private messageHistory = new Map<string, MessageHistoryEntry[]>() // Legacy: sessionId -> messages
+  private streamMessageHistory = new Map<string, MessageHistoryEntry[]>() // streamKey -> messages
   private tokenToSession = new Map<string, string>() // tokenHash -> sessionId
   private maxMessages: number
 
@@ -17,14 +18,27 @@ export class MemorySessionStore implements SessionStore {
     this.maxMessages = maxMessages
   }
 
+  private getStreamKey (sessionId: string, streamId: string): string {
+    return `${sessionId}:${streamId}`
+  }
+
   async create (metadata: SessionMetadata): Promise<void> {
-    this.sessions.set(metadata.id, { ...metadata })
+    const sessionData = {
+      ...metadata,
+      streams: new Map(metadata.streams || [])
+    }
+    this.sessions.set(metadata.id, sessionData)
     this.messageHistory.set(metadata.id, [])
   }
 
   async get (sessionId: string): Promise<SessionMetadata | null> {
     const session = this.sessions.get(sessionId)
-    return session ? { ...session } : null
+    if (!session) return null
+
+    return {
+      ...session,
+      streams: new Map(session.streams)
+    }
   }
 
   async delete (sessionId: string): Promise<void> {
@@ -34,6 +48,13 @@ export class MemorySessionStore implements SessionStore {
       this.tokenToSession.delete(session.authorization.tokenHash)
     }
 
+    // Clean up all stream message histories for this session
+    for (const [key] of this.streamMessageHistory.entries()) {
+      if (key.startsWith(`${sessionId}:`)) {
+        this.streamMessageHistory.delete(key)
+      }
+    }
+
     this.sessions.delete(sessionId)
     this.messageHistory.delete(sessionId)
   }
@@ -49,7 +70,106 @@ export class MemorySessionStore implements SessionStore {
     }
   }
 
-  async addMessage (sessionId: string, eventId: string, message: JSONRPCMessage): Promise<void> {
+  // Stream management methods
+  async createStream (sessionId: string, streamId: string): Promise<StreamMetadata | null> {
+    const session = this.sessions.get(sessionId)
+    if (!session) return null
+
+    const streamMetadata: StreamMetadata = {
+      id: streamId,
+      eventId: 0,
+      lastEventId: undefined,
+      createdAt: new Date(),
+      lastActivity: new Date()
+    }
+
+    session.streams.set(streamId, streamMetadata)
+    session.lastActivity = new Date()
+    this.sessions.set(sessionId, session)
+
+    return { ...streamMetadata }
+  }
+
+  async getStream (sessionId: string, streamId: string): Promise<StreamMetadata | null> {
+    const session = this.sessions.get(sessionId)
+    if (!session) return null
+
+    const stream = session.streams.get(streamId)
+    return stream ? { ...stream } : null
+  }
+
+  async deleteStream (sessionId: string, streamId: string): Promise<void> {
+    const session = this.sessions.get(sessionId)
+    if (!session) return
+
+    session.streams.delete(streamId)
+    session.lastActivity = new Date()
+    this.sessions.set(sessionId, session)
+
+    // Clean up stream message history
+    const streamKey = this.getStreamKey(sessionId, streamId)
+    this.streamMessageHistory.delete(streamKey)
+  }
+
+  async updateStreamActivity (sessionId: string, streamId: string): Promise<void> {
+    const session = this.sessions.get(sessionId)
+    if (!session) return
+
+    const stream = session.streams.get(streamId)
+    if (stream) {
+      stream.lastActivity = new Date()
+      session.lastActivity = new Date()
+      this.sessions.set(sessionId, session)
+    }
+  }
+
+  // Per-stream message history operations
+  async addMessage (sessionId: string, streamId: string, eventId: string, message: JSONRPCMessage): Promise<void> {
+    const streamKey = this.getStreamKey(sessionId, streamId)
+    let history = this.streamMessageHistory.get(streamKey)
+    if (!history) {
+      history = []
+      this.streamMessageHistory.set(streamKey, history)
+    }
+
+    history.push({ eventId, message })
+
+    // Auto-trim using constructor maxMessages
+    if (history.length > this.maxMessages) {
+      history.splice(0, history.length - this.maxMessages)
+    }
+
+    // Update stream metadata
+    const session = this.sessions.get(sessionId)
+    if (session) {
+      const stream = session.streams.get(streamId)
+      if (stream) {
+        stream.eventId = parseInt(eventId)
+        stream.lastEventId = eventId
+        stream.lastActivity = new Date()
+        session.lastActivity = new Date()
+        this.sessions.set(sessionId, session)
+      }
+    }
+  }
+
+  async getMessagesFrom (sessionId: string, streamId: string, fromEventId: string): Promise<Array<{ eventId: string, message: JSONRPCMessage }>> {
+    const streamKey = this.getStreamKey(sessionId, streamId)
+    const history = this.streamMessageHistory.get(streamKey) || []
+    const fromIndex = history.findIndex(entry => entry.eventId === fromEventId)
+
+    if (fromIndex === -1) {
+      return []
+    }
+
+    return history.slice(fromIndex + 1).map(entry => ({
+      eventId: entry.eventId,
+      message: entry.message
+    }))
+  }
+
+  // Session-level message operations (for broadcast messages and compatibility)
+  async addSessionMessage (sessionId: string, eventId: string, message: JSONRPCMessage): Promise<void> {
     let history = this.messageHistory.get(sessionId)
     if (!history) {
       history = []
@@ -66,12 +186,11 @@ export class MemorySessionStore implements SessionStore {
     // Update session metadata
     const session = this.sessions.get(sessionId)
     if (session) {
-      session.lastEventId = eventId
       session.lastActivity = new Date()
     }
   }
 
-  async getMessagesFrom (sessionId: string, fromEventId: string): Promise<Array<{ eventId: string, message: JSONRPCMessage }>> {
+  async getSessionMessagesFrom (sessionId: string, fromEventId: string): Promise<Array<{ eventId: string, message: JSONRPCMessage }>> {
     const history = this.messageHistory.get(sessionId) || []
     const fromIndex = history.findIndex(entry => entry.eventId === fromEventId)
 
diff --git a/src/stores/redis-session-store.ts b/src/stores/redis-session-store.ts
index d444a1d..9a3cbf0 100644
--- a/src/stores/redis-session-store.ts
+++ b/src/stores/redis-session-store.ts
@@ -1,6 +1,6 @@
 import type { Redis } from 'ioredis'
 import type { JSONRPCMessage } from '../schema.ts'
-import type { SessionStore, SessionMetadata } from './session-store.ts'
+import type { SessionStore, SessionMetadata, StreamMetadata } from './session-store.ts'
 import type { AuthorizationContext, TokenRefreshInfo } from '../types/auth-types.ts'
 
 export class RedisSessionStore implements SessionStore {
@@ -12,12 +12,18 @@ export class RedisSessionStore implements SessionStore {
     this.maxMessages = options.maxMessages || 100
   }
 
+  private getStreamKey (sessionId: string, streamId: string): string {
+    return `session:${sessionId}:stream:${streamId}`
+  }
+
+  private getStreamHistoryKey (sessionId: string, streamId: string): string {
+    return `session:${sessionId}:stream:${streamId}:history`
+  }
+
   async create (metadata: SessionMetadata): Promise<void> {
     const sessionKey = `session:${metadata.id}`
     const sessionData: Record<string, string> = {
       id: metadata.id,
-      eventId: metadata.eventId.toString(),
-      lastEventId: metadata.lastEventId || '',
       createdAt: metadata.createdAt.toISOString(),
       lastActivity: metadata.lastActivity.toISOString()
     }
@@ -33,11 +39,32 @@ export class RedisSessionStore implements SessionStore {
       sessionData.authSession = JSON.stringify(metadata.authSession)
     }
 
+    // Store stream metadata
+    if (metadata.streams && metadata.streams.size > 0) {
+      const streamsArray: Array<[string, StreamMetadata]> = Array.from(metadata.streams.entries())
+      sessionData.streams = JSON.stringify(streamsArray)
+    } else {
+      sessionData.streams = JSON.stringify([])
+    }
+
     await this.redis.hset(sessionKey, sessionData)
 
     // Set session expiration to 1 hour
     await this.redis.expire(sessionKey, 3600)
 
+    // Create stream metadata for each stream
+    for (const [streamId, streamMeta] of (metadata.streams || new Map())) {
+      const streamKey = this.getStreamKey(metadata.id, streamId)
+      await this.redis.hset(streamKey, {
+        id: streamMeta.id,
+        eventId: streamMeta.eventId.toString(),
+        lastEventId: streamMeta.lastEventId || '',
+        createdAt: streamMeta.createdAt.toISOString(),
+        lastActivity: streamMeta.lastActivity.toISOString()
+      })
+      await this.redis.expire(streamKey, 3600)
+    }
+
     // Add token mapping if present
     if (metadata.authorization?.tokenHash) {
       await this.addTokenMapping(metadata.authorization.tokenHash, metadata.id)
@@ -54,10 +81,9 @@ export class RedisSessionStore implements SessionStore {
 
     const metadata: SessionMetadata = {
       id: result.id,
-      eventId: parseInt(result.eventId, 10),
-      lastEventId: result.lastEventId || undefined,
       createdAt: new Date(result.createdAt),
-      lastActivity: new Date(result.lastActivity)
+      lastActivity: new Date(result.lastActivity),
+      streams: new Map()
     }
 
     // Parse authorization context if present
@@ -85,6 +111,17 @@ export class RedisSessionStore implements SessionStore {
       }
     }
 
+    // Parse streams data
+    if (result.streams) {
+      try {
+        const streamsArray: Array<[string, StreamMetadata]> = JSON.parse(result.streams)
+        metadata.streams = new Map(streamsArray)
+      } catch (error) {
+        // Ignore parsing errors for streams, use empty map
+        metadata.streams = new Map()
+      }
+    }
+
     return metadata
   }
 
@@ -92,12 +129,31 @@ export class RedisSessionStore implements SessionStore {
     const sessionKey = `session:${sessionId}`
     const historyKey = `session:${sessionId}:history`
 
-    // Get session to clean up token mappings
+    // Get session to clean up token mappings and streams
     const session = await this.get(sessionId)
     if (session?.authorization?.tokenHash) {
       await this.removeTokenMapping(session.authorization.tokenHash)
     }
 
+    // Clean up all streams for this session
+    if (session?.streams) {
+      for (const streamId of session.streams.keys()) {
+        const streamKey = this.getStreamKey(sessionId, streamId)
+        const streamHistoryKey = this.getStreamHistoryKey(sessionId, streamId)
+        await this.redis.del(streamKey, streamHistoryKey)
+      }
+    }
+
+    // Also scan for any missed stream keys
+    let cursor = '0'
+    do {
+      const [nextCursor, keys] = await this.redis.scan(cursor, 'MATCH', `session:${sessionId}:stream:*`, 'COUNT', 100)
+      cursor = nextCursor
+      if (keys.length > 0) {
+        await this.redis.del(...keys)
+      }
+    } while (cursor !== '0')
+
     await this.redis.del(sessionKey, historyKey)
   }
 
@@ -117,9 +173,167 @@ export class RedisSessionStore implements SessionStore {
         }
       }
     } while (cursor !== '0')
+
+    // Also clean up orphaned stream keys
+    cursor = '0'
+    do {
+      const [nextCursor, keys] = await this.redis.scan(cursor, 'MATCH', 'session:*:stream:*', 'COUNT', 100)
+      cursor = nextCursor
+      for (const key of keys) {
+        const parts = key.split(':')
+        if (parts.length >= 2) {
+          const sessionId = parts[1]
+          const sessionKey = `session:${sessionId}`
+          const exists = await this.redis.exists(sessionKey)
+          if (!exists) {
+            await this.redis.del(key)
+          }
+        }
+      }
+    } while (cursor !== '0')
   }
 
-  async addMessage (sessionId: string, eventId: string, message: JSONRPCMessage): Promise<void> {
+  // Stream management methods
+  async createStream (sessionId: string, streamId: string): Promise<StreamMetadata | null> {
+    const session = await this.get(sessionId)
+    if (!session) return null
+
+    const streamMetadata: StreamMetadata = {
+      id: streamId,
+      eventId: 0,
+      lastEventId: undefined,
+      createdAt: new Date(),
+      lastActivity: new Date()
+    }
+
+    // Add stream to session
+    session.streams.set(streamId, streamMetadata)
+    session.lastActivity = new Date()
+
+    // Update session with new stream data
+    const sessionKey = `session:${sessionId}`
+    const streamsArray: Array<[string, StreamMetadata]> = Array.from(session.streams.entries())
+    await this.redis.hset(sessionKey, {
+      streams: JSON.stringify(streamsArray),
+      lastActivity: session.lastActivity.toISOString()
+    })
+
+    // Create stream metadata in Redis
+    const streamKey = this.getStreamKey(sessionId, streamId)
+    await this.redis.hset(streamKey, {
+      id: streamMetadata.id,
+      eventId: streamMetadata.eventId.toString(),
+      lastEventId: streamMetadata.lastEventId || '',
+      createdAt: streamMetadata.createdAt.toISOString(),
+      lastActivity: streamMetadata.lastActivity.toISOString()
+    })
+    await this.redis.expire(streamKey, 3600)
+
+    return { ...streamMetadata }
+  }
+
+  async getStream (sessionId: string, streamId: string): Promise<StreamMetadata | null> {
+    const streamKey = this.getStreamKey(sessionId, streamId)
+    const result = await this.redis.hgetall(streamKey)
+
+    if (!result.id) {
+      return null
+    }
+
+    return {
+      id: result.id,
+      eventId: parseInt(result.eventId, 10),
+      lastEventId: result.lastEventId || undefined,
+      createdAt: new Date(result.createdAt),
+      lastActivity: new Date(result.lastActivity)
+    }
+  }
+
+  async deleteStream (sessionId: string, streamId: string): Promise<void> {
+    const session = await this.get(sessionId)
+    if (!session) return
+
+    // Remove stream from session
+    session.streams.delete(streamId)
+    session.lastActivity = new Date()
+
+    // Update session with new stream data
+    const sessionKey = `session:${sessionId}`
+    const streamsArray: Array<[string, StreamMetadata]> = Array.from(session.streams.entries())
+    await this.redis.hset(sessionKey, {
+      streams: JSON.stringify(streamsArray),
+      lastActivity: session.lastActivity.toISOString()
+    })
+
+    // Delete stream metadata and history
+    const streamKey = this.getStreamKey(sessionId, streamId)
+    const streamHistoryKey = this.getStreamHistoryKey(sessionId, streamId)
+    await this.redis.del(streamKey, streamHistoryKey)
+  }
+
+  async updateStreamActivity (sessionId: string, streamId: string): Promise<void> {
+    const streamKey = this.getStreamKey(sessionId, streamId)
+    const sessionKey = `session:${sessionId}`
+    const now = new Date().toISOString()
+
+    const pipeline = this.redis.pipeline()
+    pipeline.hset(streamKey, 'lastActivity', now)
+    pipeline.expire(streamKey, 3600)
+    pipeline.hset(sessionKey, 'lastActivity', now)
+    pipeline.expire(sessionKey, 3600)
+    await pipeline.exec()
+  }
+
+  // Per-stream message history operations
+  async addMessage (sessionId: string, streamId: string, eventId: string, message: JSONRPCMessage): Promise<void> {
+    const historyKey = this.getStreamHistoryKey(sessionId, streamId)
+    const streamKey = this.getStreamKey(sessionId, streamId)
+
+    // Use Redis pipeline for atomic operations
+    const pipeline = this.redis.pipeline()
+
+    // Add message to Redis stream
+    pipeline.xadd(historyKey, `${eventId}-0`, 'message', JSON.stringify(message))
+
+    // Trim to max messages (exact trimming)
+    pipeline.xtrim(historyKey, 'MAXLEN', this.maxMessages)
+
+    // Update stream metadata
+    pipeline.hset(streamKey, {
+      eventId,
+      lastEventId: eventId,
+      lastActivity: new Date().toISOString()
+    })
+
+    // Reset stream expiration
+    pipeline.expire(streamKey, 3600)
+
+    // Update session activity
+    const sessionKey = `session:${sessionId}`
+    pipeline.hset(sessionKey, 'lastActivity', new Date().toISOString())
+    pipeline.expire(sessionKey, 3600)
+
+    await pipeline.exec()
+  }
+
+  async getMessagesFrom (sessionId: string, streamId: string, fromEventId: string): Promise<Array<{ eventId: string, message: JSONRPCMessage }>> {
+    const historyKey = this.getStreamHistoryKey(sessionId, streamId)
+
+    try {
+      const results = await this.redis.xrange(historyKey, `(${fromEventId}-0`, '+')
+
+      return results.map(([id, fields]: [string, string[]]) => ({
+        eventId: id.split('-')[0],
+        message: JSON.parse(fields[1])
+      }))
+    } catch (error) {
+      // If stream doesn't exist, return empty array
+      return []
+    }
+  }
+
+  // Session-level message operations (for broadcast messages and compatibility)
+  async addSessionMessage (sessionId: string, eventId: string, message: JSONRPCMessage): Promise<void> {
     const historyKey = `session:${sessionId}:history`
     const sessionKey = `session:${sessionId}`
 
@@ -134,8 +348,6 @@ export class RedisSessionStore implements SessionStore {
 
     // Update session metadata
     pipeline.hset(sessionKey, {
-      eventId,
-      lastEventId: eventId,
       lastActivity: new Date().toISOString()
     })
 
@@ -145,7 +357,7 @@ export class RedisSessionStore implements SessionStore {
     await pipeline.exec()
   }
 
-  async getMessagesFrom (sessionId: string, fromEventId: string): Promise<Array<{ eventId: string, message: JSONRPCMessage }>> {
+  async getSessionMessagesFrom (sessionId: string, fromEventId: string): Promise<Array<{ eventId: string, message: JSONRPCMessage }>> {
     const historyKey = `session:${sessionId}:history`
 
     try {
diff --git a/src/stores/session-store.ts b/src/stores/session-store.ts
index 774803b..8b6125d 100644
--- a/src/stores/session-store.ts
+++ b/src/stores/session-store.ts
@@ -1,17 +1,26 @@
 import type { JSONRPCMessage } from '../schema.ts'
 import type { AuthorizationContext, TokenRefreshInfo } from '../types/auth-types.ts'
 
-export interface SessionMetadata {
+export interface StreamMetadata {
   id: string
   eventId: number
   lastEventId?: string
   createdAt: Date
   lastActivity: Date
+}
+
+export interface SessionMetadata {
+  id: string
+  createdAt: Date
+  lastActivity: Date
   authSession?: any // OAuth session data (legacy - for Phase 2 compatibility)
 
   // Enhanced authorization context
   authorization?: AuthorizationContext
   tokenRefresh?: TokenRefreshInfo
+
+  // Per-stream tracking - maps streamId to stream metadata
+  streams: Map<string, StreamMetadata>
 }
 
 export interface SessionStore {
@@ -20,9 +29,19 @@ export interface SessionStore {
   delete(sessionId: string): Promise<void>
   cleanup(): Promise<void>
 
-  // Message history operations
-  addMessage(sessionId: string, eventId: string, message: JSONRPCMessage): Promise<void>
-  getMessagesFrom(sessionId: string, fromEventId: string): Promise<Array<{ eventId: string, message: JSONRPCMessage }>>
+  // Stream management within sessions
+  createStream(sessionId: string, streamId: string): Promise<StreamMetadata | null>
+  getStream(sessionId: string, streamId: string): Promise<StreamMetadata | null>
+  deleteStream(sessionId: string, streamId: string): Promise<void>
+  updateStreamActivity(sessionId: string, streamId: string): Promise<void>
+
+  // Per-stream message history operations
+  addMessage(sessionId: string, streamId: string, eventId: string, message: JSONRPCMessage): Promise<void>
+  getMessagesFrom(sessionId: string, streamId: string, fromEventId: string): Promise<Array<{ eventId: string, message: JSONRPCMessage }>>
+
+  // Legacy message operations (for backwards compatibility)
+  addSessionMessage(sessionId: string, eventId: string, message: JSONRPCMessage): Promise<void>
+  getSessionMessagesFrom(sessionId: string, fromEventId: string): Promise<Array<{ eventId: string, message: JSONRPCMessage }>>
 
   // Token-to-session mapping operations
   getSessionByTokenHash(tokenHash: string): Promise<SessionMetadata | null>
diff --git a/src/types.ts b/src/types.ts
index 4ca6a9e..4c11077 100644
--- a/src/types.ts
+++ b/src/types.ts
@@ -28,7 +28,7 @@ export interface HandlerContext {
 export type ToolHandler<TSchema extends TObject = TObject> = (
   params: Static<TSchema>,
   context: HandlerContext
-) => Promise<CallToolResult> | CallToolResult
+) => Promise<CallToolResult> | CallToolResult | AsyncGenerator<CallToolResult, CallToolResult, unknown>
 
 export type ResourceHandler<TUriSchema extends TSchema = TString> = (
   uri: Static<TUriSchema>,
@@ -111,7 +111,7 @@ declare module 'fastify' {
 }
 
 // Unsafe handler types for backward compatibility
-export type UnsafeToolHandler = (params: any, context: HandlerContext) => Promise<CallToolResult> | CallToolResult
+export type UnsafeToolHandler = (params: any, context: HandlerContext) => Promise<CallToolResult> | CallToolResult | AsyncGenerator<CallToolResult, CallToolResult, unknown>
 export type UnsafeResourceHandler = (uri: string, context: HandlerContext) => Promise<ReadResourceResult> | ReadResourceResult
 export type UnsafePromptHandler = (name: string, args: any, context: HandlerContext) => Promise<GetPromptResult> | GetPromptResult
 
diff --git a/test/async-iterator-streaming.test.ts b/test/async-iterator-streaming.test.ts
new file mode 100644
index 0000000..fe86eae
--- /dev/null
+++ b/test/async-iterator-streaming.test.ts
@@ -0,0 +1,404 @@
+import { test, describe } from 'node:test'
+import { strict as assert } from 'node:assert'
+import Fastify from 'fastify'
+import { request, Agent, setGlobalDispatcher } from 'undici'
+import mcpPlugin from '../src/index.ts'
+
+setGlobalDispatcher(new Agent({
+  keepAliveTimeout: 10,
+  keepAliveMaxTimeout: 10
+}))
+
+describe('Async Iterator Streaming Tests', () => {
+  test('should return immediate JSON response for non-async-iterator tool results', async (t) => {
+    const app = Fastify({ logger: false })
+
+    t.after(async () => {
+      await app.close()
+    })
+
+    await app.register(mcpPlugin, {
+      serverInfo: { name: 'test-server', version: '1.0.0' },
+      enableSSE: true
+    })
+
+    // Regular tool that returns immediate result
+    app.mcpAddTool({
+      name: 'immediate_tool',
+      description: 'Tool that returns immediate result',
+      inputSchema: {
+        type: 'object',
+        properties: {
+          value: { type: 'string' }
+        },
+        required: ['value']
+      }
+    }, async (params) => {
+      return {
+        content: [{ type: 'text', text: `Immediate result: ${params.value}` }]
+      }
+    })
+
+    await app.ready()
+
+    const response = await app.inject({
+      method: 'POST',
+      url: '/mcp',
+      headers: {
+        'content-type': 'application/json'
+      },
+      payload: {
+        jsonrpc: '2.0',
+        id: 1,
+        method: 'tools/call',
+        params: {
+          name: 'immediate_tool',
+          arguments: { value: 'test' }
+        }
+      }
+    })
+
+    assert.strictEqual(response.statusCode, 200)
+    assert.strictEqual(response.headers['content-type'], 'application/json; charset=utf-8')
+
+    const result = JSON.parse(response.payload)
+    assert.strictEqual(result.jsonrpc, '2.0')
+    assert.strictEqual(result.id, 1)
+    assert.deepStrictEqual(result.result.content, [{ type: 'text', text: 'Immediate result: test' }])
+  })
+
+  test('should return text/event-stream for async iterator tool results', async (t) => {
+    const app = Fastify({ logger: false })
+
+    t.after(async () => {
+      await app.close()
+    })
+
+    await app.register(mcpPlugin, {
+      serverInfo: { name: 'test-server', version: '1.0.0' },
+      enableSSE: true
+    })
+
+    // Async generator tool that returns streaming results
+    app.mcpAddTool({
+      name: 'streaming_tool',
+      description: 'Tool that returns async iterator results',
+      inputSchema: {
+        type: 'object',
+        properties: {
+          count: { type: 'number' }
+        },
+        required: ['count']
+      }
+    }, async function * (params) {
+      for (let i = 1; i <= params.count; i++) {
+        yield {
+          content: [{ type: 'text', text: `Chunk ${i}` }]
+        }
+        // Small delay to simulate async work
+        await new Promise(resolve => setTimeout(resolve, 10))
+      }
+      return {
+        content: [{ type: 'text', text: 'Final result' }]
+      }
+    })
+
+    await app.ready()
+    const baseUrl = await app.listen({ port: 0 })
+
+    const response = await request(`${baseUrl}/mcp`, {
+      method: 'POST',
+      headers: {
+        'content-type': 'application/json'
+      },
+      body: JSON.stringify({
+        jsonrpc: '2.0',
+        id: 2,
+        method: 'tools/call',
+        params: {
+          name: 'streaming_tool',
+          arguments: { count: 3 }
+        }
+      })
+    })
+
+    assert.strictEqual(response.statusCode, 200)
+    assert.strictEqual(response.headers['content-type'], 'text/event-stream')
+
+    // Parse SSE events from the response
+    const responseText = await response.body.text()
+    const events = parseSSEEvents(responseText)
+
+    // Should have 4 events: 3 chunks + 1 final result
+    assert.strictEqual(events.length, 4)
+
+    // Check the content of each event
+    assert.deepStrictEqual(JSON.parse(events[0].data), {
+      jsonrpc: '2.0',
+      id: 2,
+      result: { content: [{ type: 'text', text: 'Chunk 1' }] }
+    })
+
+    assert.deepStrictEqual(JSON.parse(events[1].data), {
+      jsonrpc: '2.0',
+      id: 2,
+      result: { content: [{ type: 'text', text: 'Chunk 2' }] }
+    })
+
+    assert.deepStrictEqual(JSON.parse(events[2].data), {
+      jsonrpc: '2.0',
+      id: 2,
+      result: { content: [{ type: 'text', text: 'Chunk 3' }] }
+    })
+
+    assert.deepStrictEqual(JSON.parse(events[3].data), {
+      jsonrpc: '2.0',
+      id: 2,
+      result: { content: [{ type: 'text', text: 'Final result' }] }
+    })
+  })
+
+  test('should handle errors during streaming gracefully', async (t) => {
+    const app = Fastify({ logger: false })
+
+    t.after(async () => {
+      await app.close()
+    })
+
+    await app.register(mcpPlugin, {
+      serverInfo: { name: 'test-server', version: '1.0.0' },
+      enableSSE: true
+    })
+
+    // Async generator tool that throws an error
+    app.mcpAddTool({
+      name: 'error_tool',
+      description: 'Tool that errors during streaming',
+      inputSchema: {
+        type: 'object',
+        properties: {
+          errorAt: { type: 'number' }
+        },
+        required: ['errorAt']
+      }
+    }, async function * (params) {
+      for (let i = 1; i <= 5; i++) {
+        if (i === params.errorAt) {
+          throw new Error(`Error at chunk ${i}`)
+        }
+        yield {
+          content: [{ type: 'text', text: `Chunk ${i}` }]
+        }
+        await new Promise(resolve => setTimeout(resolve, 10))
+      }
+      return {
+        content: [{ type: 'text', text: 'Should not reach here' }]
+      }
+    })
+
+    await app.ready()
+    const baseUrl = await app.listen({ port: 0 })
+
+    const response = await request(`${baseUrl}/mcp`, {
+      method: 'POST',
+      headers: {
+        'content-type': 'application/json'
+      },
+      body: JSON.stringify({
+        jsonrpc: '2.0',
+        id: 3,
+        method: 'tools/call',
+        params: {
+          name: 'error_tool',
+          arguments: { errorAt: 3 }
+        }
+      })
+    })
+
+    assert.strictEqual(response.statusCode, 200)
+    assert.strictEqual(response.headers['content-type'], 'text/event-stream')
+
+    const responseText = await response.body.text()
+    const events = parseSSEEvents(responseText)
+
+    // Should have 3 events: 2 successful chunks + 1 error event
+    assert.strictEqual(events.length, 3)
+
+    // Check successful chunks
+    assert.deepStrictEqual(JSON.parse(events[0].data), {
+      jsonrpc: '2.0',
+      id: 3,
+      result: { content: [{ type: 'text', text: 'Chunk 1' }] }
+    })
+
+    assert.deepStrictEqual(JSON.parse(events[1].data), {
+      jsonrpc: '2.0',
+      id: 3,
+      result: { content: [{ type: 'text', text: 'Chunk 2' }] }
+    })
+
+    // Check error event
+    const errorEvent = JSON.parse(events[2].data)
+    assert.strictEqual(errorEvent.jsonrpc, '2.0')
+    assert.strictEqual(errorEvent.id, 3)
+    assert.ok(errorEvent.error)
+    assert.ok(errorEvent.error.message.includes('Error at chunk 3'))
+  })
+
+  test('should use per-session event ID system for streaming', async (t) => {
+    const app = Fastify({ logger: false })
+
+    t.after(async () => {
+      await app.close()
+    })
+
+    await app.register(mcpPlugin, {
+      serverInfo: { name: 'test-server', version: '1.0.0' },
+      enableSSE: true
+    })
+
+    // Tool that returns a few chunks
+    app.mcpAddTool({
+      name: 'event_id_tool',
+      description: 'Tool for testing event IDs',
+      inputSchema: {
+        type: 'object',
+        properties: {
+          chunks: { type: 'number' }
+        },
+        required: ['chunks']
+      }
+    }, async function * (params) {
+      for (let i = 1; i <= params.chunks; i++) {
+        yield {
+          content: [{ type: 'text', text: `Event ${i}` }]
+        }
+        await new Promise(resolve => setTimeout(resolve, 5))
+      }
+      return {
+        content: [{ type: 'text', text: 'Final event' }]
+      }
+    })
+
+    await app.ready()
+    const baseUrl = await app.listen({ port: 0 })
+
+    const response = await request(`${baseUrl}/mcp`, {
+      method: 'POST',
+      headers: {
+        'content-type': 'application/json'
+      },
+      body: JSON.stringify({
+        jsonrpc: '2.0',
+        id: 4,
+        method: 'tools/call',
+        params: {
+          name: 'event_id_tool',
+          arguments: { chunks: 3 }
+        }
+      })
+    })
+
+    assert.strictEqual(response.statusCode, 200)
+    assert.strictEqual(response.headers['content-type'], 'text/event-stream')
+
+    const responseText = await response.body.text()
+    const events = parseSSEEvents(responseText)
+    assert.strictEqual(events.length, 4) // 3 yielded + 1 final return
+
+    // Check that event IDs increment properly
+    assert.strictEqual(events[0].id, '1')
+    assert.strictEqual(events[1].id, '2')
+    assert.strictEqual(events[2].id, '3')
+    assert.strictEqual(events[3].id, '4')
+  })
+
+  test('should handle async iterator that returns no values', async (t) => {
+    const app = Fastify({ logger: false })
+
+    t.after(async () => {
+      await app.close()
+    })
+
+    await app.register(mcpPlugin, {
+      serverInfo: { name: 'test-server', version: '1.0.0' },
+      enableSSE: true
+    })
+
+    // Empty async generator
+    app.mcpAddTool({
+      name: 'empty_tool',
+      description: 'Tool that returns empty iterator',
+      inputSchema: {
+        type: 'object',
+        properties: {},
+        additionalProperties: false
+      }
+    }, async function * () {
+      // Empty generator - no yields
+      return {
+        content: [{ type: 'text', text: 'Empty result' }]
+      }
+    })
+
+    await app.ready()
+    const baseUrl = await app.listen({ port: 0 })
+
+    const response = await request(`${baseUrl}/mcp`, {
+      method: 'POST',
+      headers: {
+        'content-type': 'application/json'
+      },
+      body: JSON.stringify({
+        jsonrpc: '2.0',
+        id: 5,
+        method: 'tools/call',
+        params: {
+          name: 'empty_tool',
+          arguments: {}
+        }
+      })
+    })
+
+    assert.strictEqual(response.statusCode, 200)
+    assert.strictEqual(response.headers['content-type'], 'text/event-stream')
+
+    const responseText = await response.body.text()
+    const events = parseSSEEvents(responseText)
+
+    // Should have 1 event with the final return value
+    assert.strictEqual(events.length, 1)
+    assert.deepStrictEqual(JSON.parse(events[0].data), {
+      jsonrpc: '2.0',
+      id: 5,
+      result: { content: [{ type: 'text', text: 'Empty result' }] }
+    })
+  })
+})
+
+// Helper function to parse Server-Sent Events from raw response
+function parseSSEEvents (payload: string | undefined): Array<{ id?: string; data: string; event?: string }> {
+  if (!payload) return []
+
+  const events: Array<{ id?: string; data: string; event?: string }> = []
+  const lines = payload.split('\n')
+  let currentEvent: { id?: string; data: string; event?: string } = { data: '' }
+
+  for (const line of lines) {
+    if (line.startsWith('id: ')) {
+      currentEvent.id = line.slice(4)
+    } else if (line.startsWith('event: ')) {
+      currentEvent.event = line.slice(7)
+    } else if (line.startsWith('data: ')) {
+      currentEvent.data = line.slice(6)
+    } else if (line === '') {
+      // Empty line indicates end of event
+      if (currentEvent.data) {
+        events.push({ ...currentEvent })
+      }
+      currentEvent = { data: '' }
+    }
+  }
+
+  return events
+}
diff --git a/test/per-stream-event-ids.test.ts b/test/per-stream-event-ids.test.ts
new file mode 100644
index 0000000..38d08fa
--- /dev/null
+++ b/test/per-stream-event-ids.test.ts
@@ -0,0 +1,141 @@
+import { test } from 'node:test'
+import { strict as assert } from 'node:assert'
+import Fastify from 'fastify'
+import mcpPlugin from '../src/index.ts'
+
+/**
+ * Per-Stream Event ID Tests
+ *
+ * According to MCP transport specification line 169:
+ * "These event IDs should be assigned by servers on a per-stream basis, to
+ * act as a cursor within that particular stream."
+ *
+ * This test suite verifies:
+ * 1. Event IDs are assigned on a per-stream basis (not per-session)
+ * 2. Each SSE stream has its own event ID sequence starting from 1
+ * 3. Last-Event-ID header works per-stream for reconnection
+ * 4. Stream IDs are unique within a session
+ * 5. Message storage is organized by stream, not just session
+ */
+
+test('Each SSE stream should have independent event ID sequences', async (t) => {
+  const app = Fastify({ logger: false })
+
+  t.after(async () => {
+    await app.close()
+  })
+
+  // Register MCP plugin with SSE enabled
+  await app.register(mcpPlugin, {
+    serverInfo: {
+      name: 'test-server',
+      version: '1.0.0'
+    },
+    enableSSE: true
+  })
+
+  await app.listen({ port: 0 })
+  const address = app.server.address()
+  if (!address || typeof address !== 'object') {
+    throw new Error('Failed to start test server')
+  }
+
+  // Initialize session
+  const initResponse = await app.inject({
+    method: 'POST',
+    url: '/mcp',
+    headers: {
+      'Content-Type': 'application/json',
+      Accept: 'application/json'
+    },
+    payload: JSON.stringify({
+      jsonrpc: '2.0',
+      id: 1,
+      method: 'initialize',
+      params: {
+        protocolVersion: '2025-06-18',
+        capabilities: {},
+        clientInfo: {
+          name: 'test-client',
+          version: '1.0.0'
+        }
+      }
+    })
+  })
+
+  const sessionId = initResponse.headers['mcp-session-id'] as string
+  assert.ok(sessionId, 'Session ID should be provided')
+
+  // Create first SSE stream
+  const stream1Response = await app.inject({
+    method: 'GET',
+    url: '/mcp',
+    headers: {
+      Accept: 'text/event-stream',
+      'mcp-session-id': sessionId
+    },
+    payloadAsStream: true
+  })
+
+  assert.strictEqual(stream1Response.statusCode, 200)
+  assert.strictEqual(stream1Response.headers['content-type'], 'text/event-stream')
+
+  // Verify stream1 got a unique stream ID
+  const stream1Id = stream1Response.headers['mcp-stream-id']
+  assert.ok(stream1Id, 'Stream 1 should have a unique stream ID')
+
+  // Create second SSE stream to the same session
+  const stream2Response = await app.inject({
+    method: 'GET',
+    url: '/mcp',
+    headers: {
+      Accept: 'text/event-stream',
+      'mcp-session-id': sessionId
+    },
+    payloadAsStream: true
+  })
+
+  assert.strictEqual(stream2Response.statusCode, 200)
+  assert.strictEqual(stream2Response.headers['content-type'], 'text/event-stream')
+
+  // Verify stream2 got a unique stream ID different from stream1
+  const stream2Id = stream2Response.headers['mcp-stream-id']
+  assert.ok(stream2Id, 'Stream 2 should have a unique stream ID')
+  assert.notStrictEqual(stream1Id, stream2Id, 'Each stream should have a unique ID')
+
+  // Clean up streams
+  stream1Response.stream().destroy()
+  stream2Response.stream().destroy()
+})
+
+test('Last-Event-ID header should work for per-stream message replay', async () => {
+  // This test documents the expected per-stream Last-Event-ID behavior
+  // According to MCP spec, Last-Event-ID should work on a per-stream basis
+  // Current implementation uses per-session event IDs which breaks proper resumability
+
+  assert.ok(true, 'Test placeholder - per-stream Last-Event-ID implementation needed')
+})
+
+test('Multiple streams should not interfere with each other\'s event IDs', async () => {
+  // This test documents the requirement that each stream has independent event ID sequences
+  // According to MCP spec line 145: server MUST send each message on only one stream
+  // Current implementation broadcasts to all streams and shares event IDs
+
+  assert.ok(true, 'Test placeholder - independent stream event IDs needed')
+})
+
+test('Stream IDs should be unique within a session', async () => {
+  // This test documents the requirement for unique stream IDs within a session
+  // Stream IDs are needed to properly implement per-stream event ID sequences
+  // Current implementation doesn't generate or track individual stream IDs
+
+  assert.ok(true, 'Test placeholder - unique stream ID generation needed')
+})
+
+test('Message storage should be organized by stream, not just session', async () => {
+  // This test documents the requirement for per-stream message storage
+  // Messages should be stored with stream context for proper Last-Event-ID replay
+  // Current implementation stores messages per-session without stream differentiation
+
+  assert.ok(true, 'Test placeholder - per-stream message storage needed')
+})
diff --git a/test/redis-session-store.test.ts b/test/redis-session-store.test.ts
index 60dc139..77feb31 100644
--- a/test/redis-session-store.test.ts
+++ b/test/redis-session-store.test.ts
@@ -11,10 +11,9 @@ describe('RedisSessionStore', () => {
 
     const metadata: SessionMetadata = {
       id: 'test-session-1',
-      eventId: 1,
-      lastEventId: '1',
       createdAt: new Date('2023-01-01T00:00:00.000Z'),
-      lastActivity: new Date('2023-01-01T00:01:00.000Z')
+      lastActivity: new Date('2023-01-01T00:01:00.000Z'),
+      streams: new Map()
     }
 
     await store.create(metadata)
@@ -22,10 +21,10 @@ describe('RedisSessionStore', () => {
 
     assert.ok(retrieved)
     assert.strictEqual(retrieved.id, metadata.id)
-    assert.strictEqual(retrieved.eventId, metadata.eventId)
-    assert.strictEqual(retrieved.lastEventId, metadata.lastEventId)
     assert.deepStrictEqual(retrieved.createdAt, metadata.createdAt)
     assert.deepStrictEqual(retrieved.lastActivity, metadata.lastActivity)
+    assert.ok(retrieved.streams instanceof Map)
+    assert.strictEqual(retrieved.streams.size, 0)
   })
 
   testWithRedis('should return null for non-existent session', async (redis) => {
@@ -40,9 +39,9 @@ describe('RedisSessionStore', () => {
 
     const metadata: SessionMetadata = {
       id: 'test-session-2',
-      eventId: 1,
       createdAt: new Date(),
-      lastActivity: new Date()
+      lastActivity: new Date(),
+      streams: new Map()
     }
 
     await store.create(metadata)
@@ -53,7 +52,7 @@ describe('RedisSessionStore', () => {
       method: 'test',
       id: 1
     }
-    await store.addMessage('test-session-2', '1', message)
+    await store.addSessionMessage('test-session-2', '1', message)
 
     // Verify session exists
     const before = await store.get('test-session-2')
@@ -67,7 +66,7 @@ describe('RedisSessionStore', () => {
     assert.strictEqual(after, null)
 
     // Verify history is deleted
-    const history = await store.getMessagesFrom('test-session-2', '0')
+    const history = await store.getSessionMessagesFrom('test-session-2', '0')
     assert.strictEqual(history.length, 0)
   })
 
@@ -76,9 +75,10 @@ describe('RedisSessionStore', () => {
 
     const metadata: SessionMetadata = {
       id: 'test-session-3',
-      eventId: 0,
+
       createdAt: new Date(),
-      lastActivity: new Date()
+      lastActivity: new Date(),
+      streams: new Map()
     }
 
     await store.create(metadata)
@@ -95,16 +95,15 @@ describe('RedisSessionStore', () => {
       id: 2
     }
 
-    await store.addMessage('test-session-3', '1', message1)
-    await store.addMessage('test-session-3', '2', message2)
+    await store.addSessionMessage('test-session-3', '1', message1)
+    await store.addSessionMessage('test-session-3', '2', message2)
 
     // Check updated session metadata
     const updatedSession = await store.get('test-session-3')
     assert.ok(updatedSession)
-    assert.strictEqual(updatedSession.lastEventId, '2')
 
     // Check message history
-    const history = await store.getMessagesFrom('test-session-3', '0')
+    const history = await store.getSessionMessagesFrom('test-session-3', '0')
     assert.strictEqual(history.length, 2)
     assert.strictEqual(history[0].eventId, '1')
     assert.deepStrictEqual(history[0].message, message1)
@@ -117,9 +116,10 @@ describe('RedisSessionStore', () => {
 
     const metadata: SessionMetadata = {
       id: 'test-session-4',
-      eventId: 0,
+
       createdAt: new Date(),
-      lastActivity: new Date()
+      lastActivity: new Date(),
+      streams: new Map()
     }
 
     await store.create(metadata)
@@ -131,11 +131,11 @@ describe('RedisSessionStore', () => {
     ]
 
     for (let i = 0; i < messages.length; i++) {
-      await store.addMessage('test-session-4', (i + 1).toString(), messages[i])
+      await store.addSessionMessage('test-session-4', (i + 1).toString(), messages[i])
     }
 
     // Get messages from event ID 1 (should return events 2 and 3)
-    const history = await store.getMessagesFrom('test-session-4', '1')
+    const history = await store.getSessionMessagesFrom('test-session-4', '1')
     assert.strictEqual(history.length, 2)
     assert.strictEqual(history[0].eventId, '2')
     assert.deepStrictEqual(history[0].message, messages[1])
@@ -148,9 +148,10 @@ describe('RedisSessionStore', () => {
 
     const metadata: SessionMetadata = {
       id: 'test-session-5',
-      eventId: 0,
+
       createdAt: new Date(),
-      lastActivity: new Date()
+      lastActivity: new Date(),
+      streams: new Map()
     }
 
     await store.create(metadata)
@@ -162,11 +163,11 @@ describe('RedisSessionStore', () => {
         method: `test${i}`,
         id: i
       }
-      await store.addMessage('test-session-5', i.toString(), message)
+      await store.addSessionMessage('test-session-5', i.toString(), message)
     }
 
     // Should have exactly 3 messages (exact trimming)
-    const history = await store.getMessagesFrom('test-session-5', '0')
+    const history = await store.getSessionMessagesFrom('test-session-5', '0')
     assert.strictEqual(history.length, 3)
     assert.strictEqual(history[0].eventId, '3')
     assert.strictEqual(history[1].eventId, '4')
@@ -178,9 +179,10 @@ describe('RedisSessionStore', () => {
 
     const metadata: SessionMetadata = {
       id: 'test-session-6',
-      eventId: 0,
+
       createdAt: new Date(),
-      lastActivity: new Date()
+      lastActivity: new Date(),
+      streams: new Map()
     }
 
     await store.create(metadata)
@@ -191,7 +193,7 @@ describe('RedisSessionStore', () => {
       method: 'test',
       id: 1
     }
-    await store.addMessage('test-session-6', '1', message)
+    await store.addSessionMessage('test-session-6', '1', message)
 
     // Delete only the session (not the history) to simulate orphaned history
     await redis.del('session:test-session-6')
@@ -209,9 +211,10 @@ describe('RedisSessionStore', () => {
 
     const metadata: SessionMetadata = {
       id: 'test-session-7',
-      eventId: 0,
+
       createdAt: new Date(),
-      lastActivity: new Date()
+      lastActivity: new Date(),
+      streams: new Map()
     }
 
     await store.create(metadata)
@@ -226,7 +229,7 @@ describe('RedisSessionStore', () => {
       method: 'test',
       id: 1
     }
-    await store.addMessage('test-session-7', '1', message)
+    await store.addSessionMessage('test-session-7', '1', message)
 
     const newTtl = await redis.ttl('session:test-session-7')
     assert.ok(newTtl > 3500 && newTtl <= 3600)
@@ -235,7 +238,7 @@ describe('RedisSessionStore', () => {
   testWithRedis('should return empty array for non-existent message history', async (redis) => {
     const store = new RedisSessionStore({ redis, maxMessages: 100 })
 
-    const history = await store.getMessagesFrom('non-existent-session', '0')
+    const history = await store.getSessionMessagesFrom('non-existent-session', '0')
     assert.strictEqual(history.length, 0)
   })
 })
diff --git a/test/session-auth.test.ts b/test/session-auth.test.ts
index 74727f1..927105d 100644
--- a/test/session-auth.test.ts
+++ b/test/session-auth.test.ts
@@ -133,7 +133,8 @@ describe('Session-Based Authorization', () => {
         id: 'session-123',
         eventId: 0,
         createdAt: new Date(),
-        lastActivity: new Date()
+        lastActivity: new Date(),
+        streams: new Map()
       }
 
       await store.create(session)
@@ -155,7 +156,8 @@ describe('Session-Based Authorization', () => {
         id: 'session-123',
         eventId: 0,
         createdAt: new Date(),
-        lastActivity: new Date()
+        lastActivity: new Date(),
+        streams: new Map()
       }
 
       await store.create(session)
@@ -197,6 +199,7 @@ describe('Session-Based Authorization', () => {
         eventId: 0,
         createdAt: new Date(),
         lastActivity: new Date(),
+        streams: new Map(),
         authorization: {
           userId: 'user123',
           tokenHash
@@ -444,6 +447,7 @@ describe('Session-Based Authorization', () => {
         eventId: 0,
         createdAt: new Date(),
         lastActivity: new Date(),
+        streams: new Map(),
         authorization: authContext
       }
 
@@ -469,7 +473,8 @@ describe('Session-Based Authorization', () => {
         id: 'session-123',
         eventId: 0,
         createdAt: new Date(),
-        lastActivity: new Date()
+        lastActivity: new Date(),
+        streams: new Map()
       }
 
       await sessionStore.create(session)
diff --git a/test/token-refresh-coordination.test.ts b/test/token-refresh-coordination.test.ts
index 53a272c..c82934c 100644
--- a/test/token-refresh-coordination.test.ts
+++ b/test/token-refresh-coordination.test.ts
@@ -430,6 +430,7 @@ describe('Token Refresh Service Coordination', () => {
         eventId: 0,
         createdAt: new Date(),
         lastActivity: new Date(),
+        streams: new Map(),
         authorization: {
           userId: 'user123',
           tokenHash: 'old-token-hash',
diff --git a/test/token-refresh.test.ts b/test/token-refresh.test.ts
index b02d13b..44078cc 100644
--- a/test/token-refresh.test.ts
+++ b/test/token-refresh.test.ts
@@ -77,7 +77,8 @@ describe('Token Refresh Service', () => {
         id: 'session-123',
         eventId: 0,
         createdAt: new Date(),
-        lastActivity: new Date()
+        lastActivity: new Date(),
+        streams: new Map()
       }
 
       await sessionStore.create(session)
@@ -134,7 +135,8 @@ describe('Token Refresh Service', () => {
         id: 'session-123',
         eventId: 0,
         createdAt: new Date(),
-        lastActivity: new Date()
+        lastActivity: new Date(),
+        streams: new Map()
       }
 
       await sessionStore.create(session)
@@ -183,7 +185,8 @@ describe('Token Refresh Service', () => {
         id: 'session-123',
         eventId: 0,
         createdAt: new Date(),
-        lastActivity: new Date()
+        lastActivity: new Date(),
+        streams: new Map()
       }
 
       await sessionStore.create(session)
@@ -242,7 +245,8 @@ describe('Token Refresh Service', () => {
         id: 'session-123',
         eventId: 0,
         createdAt: new Date(),
-        lastActivity: new Date()
+        lastActivity: new Date(),
+        streams: new Map()
       }
 
       await sessionStore.create(session)
@@ -329,7 +333,8 @@ describe('Token Refresh Service', () => {
         id: 'session-123',
         eventId: 0,
         createdAt: new Date(),
-        lastActivity: new Date()
+        lastActivity: new Date(),
+        streams: new Map()
       }
 
       await sessionStore.create(session)