Merge pull request #596 from gargnipungarg/master

Stateful MCP python example with inferencing payloads

Author: mayoor

model-deployment/mcp-servers/stateful-python-example-server/Dockerfile

@@ -0,0 +1,10 @@
+FROM python:3.13-slim
+
+WORKDIR /app
+RUN pip install uv
+COPY mcp_simple_streamablehttp mcp_simple_streamablehttp
+COPY pyproject.toml pyproject.toml
+COPY README.md README.md
+RUN uv sync
+
+CMD ["uv", "--offline", "run", "mcp_simple_streamablehttp", "--port", "8080"]

model-deployment/mcp-servers/stateful-python-example-server/README.md

@@ -0,0 +1,60 @@
+# MCP Simple StreamableHttp stateful Server Example
+
+A simple MCP server example demonstrating the StreamableHttp transport, which enables HTTP-based communication with MCP servers using streaming.
+This example talks how to enable session id based communication in order to identify a particular user session.
+
+## Features
+
+- Uses the StreamableHTTP transport for server-client communication
+- Supports REST API operations (POST, GET, DELETE) for `/mcp` endpoint
+- Task management with anyio task groups
+- Ability to send multiple notifications over time to the client
+- Proper resource cleanup and lifespan management
+- Resumability support via InMemoryEventStore
+
+## Usage
+
+Start the server on the default or custom port:
+
+```bash
+
+# Using custom port
+uv run mcp-simple-streamablehttp --port 8080
+
+# Custom logging level
+uv run mcp-simple-streamablehttp --log-level DEBUG
+
+# Enable JSON responses instead of SSE streams
+uv run mcp-simple-streamablehttp --json-response
+```
+
+The server exposes a tool named "start-notification-stream" that accepts three arguments:
+
+- `interval`: Time between notifications in seconds (e.g., 1.0)
+- `count`: Number of notifications to send (e.g., 5)
+- `caller`: Identifier string for the caller
+
+## Resumability Support
+
+This server includes resumability support through the InMemoryEventStore. This enables clients to:
+
+- Reconnect to the server after a disconnection
+- Resume event streaming from where they left off using the Last-Event-ID header
+
+
+The server will:
+- Generate unique event IDs for each SSE message
+- Store events in memory for later replay
+- Replay missed events when a client reconnects with a Last-Event-ID header
+
+Note: The InMemoryEventStore is designed for demonstration purposes only. For production use, consider implementing a persistent storage solution.
+
+
+
+## Client
+
+You can connect to this server using an HTTP client, for now only Typescript SDK has streamable HTTP client examples or you can use [Inspector](https://github.com/modelcontextprotocol/inspector)
+
+## Limitations
+As this examples shows session ids in ephermal and in memory state, any update/restart/patching will wipe out session instances.
+Also as this is not a distributed architecture, there is no guarantee of session persistence in more than 1 nodes. Users need to implement custom, centralised session management persistence layer.

model-deployment/mcp-servers/stateful-python-example-server/inference.py

@@ -0,0 +1,136 @@
+import json
+
+import requests
+import oci
+
+MCP_SESSION_ID = "mcp-session-id"
+
+def get_auth():
+    PROFILE_NAME = 'DEFAULT'
+    SECURITY_TOKEN_FILE_KEY = 'security_token_file'
+    KEY_FILE_KEY = 'key_file'
+    config = oci.config.from_file(profile_name=PROFILE_NAME)
+    token_file = config[SECURITY_TOKEN_FILE_KEY]
+    token = None
+    with open(token_file, 'r') as f:
+      token = f.read()
+    private_key = oci.signer.load_private_key_from_file(config[KEY_FILE_KEY])
+    signer = oci.auth.signers.SecurityTokenSigner(token, private_key) 
+    return signer
+
+def initializeSession(url):
+   request_headers = {
+      "Accept-Encoding":"gzip, deflate, br, zstd", 
+      "Accept-Language":"en-GB,en-US;q=0.9,en;q=0.8", 
+      "Connection":"keep-alive", 
+      "accept":"application/json, "
+      "text/event-stream", 
+      "content-type":"application/json"
+    }
+   
+   request_body = json.dumps({
+      "method":"initialize",
+      "params": {
+        "protocolVersion":"2025-03-26",
+         "capabilities": {
+            "sampling":{},
+            "roots": {
+               "listChanged":True
+             }
+          },
+          "clientInfo": {
+             "name":"oci-cli",
+             "version":"0.14.0"
+          }
+        },
+        "jsonrpc":"2.0",
+        "id":0
+    })
+   response = requests.request("POST", url, headers=request_headers, data=request_body, auth=get_auth(), verify=False)
+   print("Initialized instance")
+   print(response.content)
+   print(response.headers)
+   return response.headers[MCP_SESSION_ID]
+
+def initializeGroup(url, mcpSessionId):
+    request_headers = {
+      "Accept-Encoding":"gzip, deflate, br, zstd", 
+      "Accept-Language":"en-GB,en-US;q=0.9,en;q=0.8", 
+      "Connection":"keep-alive", 
+      "accept":"application/json, "
+      "text/event-stream", 
+      "content-type":"application/json",
+      MCP_SESSION_ID: mcpSessionId
+    }
+   
+    request_body = json.dumps({"method":"notifications/initialized","jsonrpc":"2.0"})
+    response = requests.request("POST", url, headers=request_headers, data=request_body, auth=get_auth(), verify=False)
+    print("Initialized group")
+
+def listTools(url, mcpSessionId):
+    request_headers = {
+      "Accept-Encoding":"gzip, deflate, br, zstd", 
+      "Accept-Language":"en-GB,en-US;q=0.9,en;q=0.8", 
+      "Connection":"keep-alive", 
+      "accept":"application/json, "
+      "text/event-stream", 
+      "content-type":"application/json",
+      MCP_SESSION_ID: mcpSessionId
+    }
+   
+    request_body = json.dumps({"method":"tools/list","params":{"_meta":{"progressToken":1}},"jsonrpc":"2.0","id":1})
+    response = requests.request("POST", url, headers=request_headers, data=request_body, auth=get_auth(), verify=False)
+    print("Listing available tools")
+    print(response.content)
+
+
+def callTools(url, mcpSessionId):
+    request_headers = {
+      "Accept-Encoding":"gzip, deflate, br, zstd", 
+      "Accept-Language":"en-GB,en-US;q=0.9,en;q=0.8", 
+      "Connection":"keep-alive", 
+      "accept":"application/json, "
+      "text/event-stream", 
+      "content-type":"application/json",
+      MCP_SESSION_ID: mcpSessionId
+    }
+
+    request_body = json.dumps({"method":"tools/call","params":{"name":"start-notification-stream","arguments":{"interval":2,"count":5,"caller":"nipun"},"_meta":{"progressToken":2}},"jsonrpc":"2.0","id":2})
+
+    with requests.request("POST", url, headers=request_headers, data=request_body, auth=get_auth(), verify=False, stream=True) as response:
+      response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
+      for chunk in response.iter_content(chunk_size=8192): # 8192 is the default chunk size
+          if chunk: # Filter out keep-alive new chunks
+              print(chunk)
+
+if __name__ == "__main__":
+     predict_url = '<MODEL_DEPLOYMENT_STREAMING_URL..../predictWithResponseStream>'
+    
+     # Generate MCP session id
+     mcpSessionId = initializeSession(predict_url)
+
+     # initialise group
+     initializeGroup(predict_url, mcpSessionId)
+
+     # list tools
+     listTools(predict_url, mcpSessionId)
+
+     # call streaming tool
+     callTools(predict_url, mcpSessionId)
+
+# Sample run logs - 
+
+# > python3 inference.py 
+#Initialized instance
+#b'id: 825b5179-1ad3-425a-8dc3-95f3f369278b\r\nevent: message\r\ndata: {"jsonrpc":"2.0","id":0,"result":{"protocolVersion":"2025-03-26","capabilities":{"experimental":{},"tools":{"listChanged":false}},"serverInfo":{"name":"mcp-streamable-http-demo","version":"1.9.4.dev7+29c69e6"}}}\r\n\r\n'
+#{'date': 'Thu, 12 Jun 2025 06:01:19 GMT', 'server': 'uvicorn', 'cache-control': 'no-cache, no-transform', 'connection': 'keep-alive', 'content-type': 'text/event-stream', 'mcp-session-id': '4161e6185f934d289dbe7e61c32f3d97', 'x-accel-buffering': 'no', 'Transfer-Encoding': 'chunked'}
+#Initialized group
+#Listing available tools
+#<Response [202]>
+#b'id: a8beaa09-36f5-4943-bf2f-a1e141805a3c\r\nevent: message\r\ndata: {"method":"notifications/message","params":{"level":"info","logger":"notification_stream","data":"[1/5] Event from \'nipun\' - Use Last-Event-ID to resume if disconnected"},"jsonrpc":"2.0"}\r\n\r\n'
+#b'id: b81de9d0-b520-4714-8b07-75a8d57f1bc1\r\nevent: message\r\ndata: {"method":"notifications/message","params":{"level":"info","logger":"notification_stream","data":"[2/5] Event from \'nipun\' - Use Last-Event-ID to resume if disconnected"},"jsonrpc":"2.0"}\r\n\r\n'
+#b'id: 2ae71b1c-63d5-4f37-8830-e209e5beedf0\r\nevent: message\r\ndata: {"method":"notifications/message","params":{"level":"info","logger":"notification_stream","data":"[3/5] Event from \'nipun\' - Use Last-Event-ID to resume if disconnected"},"jsonrpc":"2.0"}\r\n\r\n'
+#b'id: bf26d789-170d-40f2-b67c-10c1f5f191cb\r\nevent: message\r\ndata: {"method":"notifications/message","params":{"level":"info","logger":"notification_stream","data":"[4/5] Event from \'nipun\' - Use Last-Event-ID to resume if disconnected"},"jsonrpc":"2.0"}\r\n\r\n'
+#b'id: b3bb3f5a-729f-4347-b64a-aad02a8a2d33\r\nevent: message\r\ndata: {"method":"notifications/message","params":{"level":"info","logger":"notification_stream","data":"[5/5] Event from \'nipun\' - Use Last-Event-ID to resume if disconnected"},"jsonrpc":"2.0"}\r\n\r\n'
+#b'id: aa0837f6-ac17-4aa7-97a6-3bc7046a1ead\r\nevent: message\r\ndata: {"jsonrpc":"2.0","id":2,"result":{"content":[{"type":"text","text":"Sent 5 notifications with 2s interval for caller: nipun"}],"isError":false}}\r\n\r\n'
+

model-deployment/mcp-servers/stateful-python-example-server/mcp_simple_streamablehttp/__init__.py

@@ -0,0 +1,4 @@
+from server import main
+
+if __name__ == "__main__":
+    main()  # type: ignore[call-arg]

model-deployment/mcp-servers/stateful-python-example-server/mcp_simple_streamablehttp/__main__.py

@@ -0,0 +1,105 @@
+"""
+In-memory event store for demonstrating resumability functionality.
+
+This is a simple implementation intended for examples and testing,
+not for production use where a persistent storage solution would be more appropriate.
+"""
+
+import logging
+from collections import deque
+from dataclasses import dataclass
+from uuid import uuid4
+
+from mcp.server.streamable_http import (
+    EventCallback,
+    EventId,
+    EventMessage,
+    EventStore,
+    StreamId,
+)
+from mcp.types import JSONRPCMessage
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class EventEntry:
+    """
+    Represents an event entry in the event store.
+    """
+
+    event_id: EventId
+    stream_id: StreamId
+    message: JSONRPCMessage
+
+
+class InMemoryEventStore(EventStore):
+    """
+    Simple in-memory implementation of the EventStore interface for resumability.
+    This is primarily intended for examples and testing, not for production use
+    where a persistent storage solution would be more appropriate.
+
+    This implementation keeps only the last N events per stream for memory efficiency.
+    """
+
+    def __init__(self, max_events_per_stream: int = 100):
+        """Initialize the event store.
+
+        Args:
+            max_events_per_stream: Maximum number of events to keep per stream
+        """
+        self.max_events_per_stream = max_events_per_stream
+        # for maintaining last N events per stream
+        self.streams: dict[StreamId, deque[EventEntry]] = {}
+        # event_id -> EventEntry for quick lookup
+        self.event_index: dict[EventId, EventEntry] = {}
+
+    async def store_event(
+        self, stream_id: StreamId, message: JSONRPCMessage
+    ) -> EventId:
+        """Stores an event with a generated event ID."""
+        event_id = str(uuid4())
+        event_entry = EventEntry(
+            event_id=event_id, stream_id=stream_id, message=message
+        )
+
+        # Get or create deque for this stream
+        if stream_id not in self.streams:
+            self.streams[stream_id] = deque(maxlen=self.max_events_per_stream)
+
+        # If deque is full, the oldest event will be automatically removed
+        # We need to remove it from the event_index as well
+        if len(self.streams[stream_id]) == self.max_events_per_stream:
+            oldest_event = self.streams[stream_id][0]
+            self.event_index.pop(oldest_event.event_id, None)
+
+        # Add new event
+        self.streams[stream_id].append(event_entry)
+        self.event_index[event_id] = event_entry
+
+        return event_id
+
+    async def replay_events_after(
+        self,
+        last_event_id: EventId,
+        send_callback: EventCallback,
+    ) -> StreamId | None:
+        """Replays events that occurred after the specified event ID."""
+        if last_event_id not in self.event_index:
+            logger.warning(f"Event ID {last_event_id} not found in store")
+            return None
+
+        # Get the stream and find events after the last one
+        last_event = self.event_index[last_event_id]
+        stream_id = last_event.stream_id
+        stream_events = self.streams.get(last_event.stream_id, deque())
+
+        # Events in deque are already in chronological order
+        found_last = False
+        for event in stream_events:
+            if found_last:
+                await send_callback(EventMessage(event.message, event.event_id))
+            elif event.event_id == last_event_id:
+                found_last = True
+
+        return stream_id