[Feature Request]: Update Streamable HTTP to fully support Virtual Servers

[crivetimihai]

🧭 Epic

Title: Streamable HTTP Virtual Server Support Parity
Goal: Update Streamable HTTP transport to properly scope tools/resources/prompts to specific virtual servers, matching SSE transport behavior.
Why now: Streamable HTTP currently exposes all global tools/resources/prompts regardless of virtual server context, breaking the virtual server isolation model. This creates inconsistent behavior between transports and violates security boundaries.
Note: The SSE transport correctly implements virtual server scoping by passing server_id context through the session registry. Streamable HTTP needs equivalent functionality.

---

🧭 Type of Feature

• Bug fix (transport inconsistency)
• Enhancement (feature parity)
• Transport improvement

---

🙋‍♂️ User Story 1 - Consistent Virtual Server Scoping

As a: MCP client developer
I want: Streamable HTTP transport to only expose tools/resources/prompts from the specific virtual server I'm connecting to
So that: behavior is consistent between SSE and Streamable HTTP transports and virtual server isolation is maintained.

✅ Acceptance Criteria

Scenario: Connect to virtual server via Streamable HTTP
Given a virtual server with ID 1 has tools [A, B] and global catalog has tools [A, B, C, D]
When I connect via Streamable HTTP to /servers/1/mcp
Then tools/list should return only tools [A, B]
And tools/list should NOT return global tools [C, D]

---

🙋‍♂️ User Story 2 - Complete MCP Method Support

As a: MCP client
I want: all standard MCP methods (resources/list, prompts/list, tools/call, etc.) to work with virtual server scoping via Streamable HTTP
So that: I have full MCP functionality with proper isolation.

✅ Acceptance Criteria

Scenario: Full MCP operations on virtual server
Given a virtual server with associated resources and prompts
When I call resources/list via Streamable HTTP
Then I receive only resources associated with that virtual server
And when I call prompts/list I receive only server-scoped prompts
And when I call tools/call it executes properly with server context

---

🙋‍♂️ User Story 3 - Transport Behavior Parity

As a: Gateway administrator
I want: identical virtual server behavior between SSE and Streamable HTTP transports
So that: clients can choose transport based on technical requirements without functional differences.

✅ Acceptance Criteria

Scenario: Transport behavior consistency
Given the same virtual server accessed via both transports
When I call tools/list via SSE transport
And I call tools/list via Streamable HTTP transport  
Then both responses contain identical tool sets
And both respect the same virtual server scoping rules

---

📐 Current Issue Analysis

Problem Identified:

# streamablehttp_transport.py - CURRENT IMPLEMENTATION
@mcp_app.list_tools()
async def list_tools() -> List[types.Tool]:
    server_id = server_id_var.get()  # ✅ Correctly extracts server_id
    
    if server_id:
        # ✅ Correctly calls server-specific method
        tools = await tool_service.list_server_tools(db, server_id)
    else:
        # ❌ Falls back to global tools 
        tools = await tool_service.list_tools(db)

Missing Implementations:

# ❌ NOT IMPLEMENTED - Missing methods
@mcp_app.list_resources()  # Missing entirely
@mcp_app.list_prompts()    # Missing entirely
@mcp_app.call_tool()       # Missing server context
@mcp_app.get_prompt()      # Missing server context
@mcp_app.read_resource()   # Missing server context

SSE Implementation (Working Correctly):

# session_registry.py - REFERENCE IMPLEMENTATION
if method == "tools/list":
    if server_id:  # ✅ Properly scoped
        tools = await tool_service.list_server_tools(db, server_id=server_id)
    else:
        tools = await tool_service.list_tools(db)
        
elif method == "resources/list":
    if server_id:  # ✅ Properly scoped
        resources = await resource_service.list_server_resources(db, server_id=server_id)
    else:
        resources = await resource_service.list_resources(db)

---

🔧 Technical Implementation Plan

Phase 1: Add Missing MCP Method Handlers

# streamablehttp_transport.py - REQUIRED ADDITIONS

@mcp_app.list_resources()
async def list_resources() -> List[types.Resource]:
    """List resources with virtual server scoping."""
    server_id = server_id_var.get()
    
    async with get_db() as db:
        if server_id:
            resources = await resource_service.list_server_resources(db, server_id)
        else:
            resources = await resource_service.list_resources(db)
    
    return [types.Resource(uri=r.uri, name=r.name, description=r.description, 
                          mimeType=r.mime_type) for r in resources]

@mcp_app.list_prompts()
async def list_prompts() -> List[types.Prompt]:
    """List prompts with virtual server scoping."""
    server_id = server_id_var.get()
    
    async with get_db() as db:
        if server_id:
            prompts = await prompt_service.list_server_prompts(db, server_id)
        else:
            prompts = await prompt_service.list_prompts(db)
    
    return [types.Prompt(name=p.name, description=p.description, 
                        arguments=p.arguments) for p in prompts]

@mcp_app.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[types.TextContent]:
    """Call tool with virtual server context."""
    server_id = server_id_var.get()
    
    async with get_db() as db:
        if server_id:
            # Verify tool belongs to this virtual server
            tools = await tool_service.list_server_tools(db, server_id)
            if not any(tool.name == name for tool in tools):
                raise ValueError(f"Tool '{name}' not available in server {server_id}")
        
        result = await tool_service.call_tool(db, name, arguments)
        return result.content

Phase 2: Context Variable Management

# Ensure server_id context is properly set for all virtual server requests
class MCPPathRewriteMiddleware:
    async def __call__(self, scope, receive, send):
        # Extract server_id from path and set context
        path = scope.get("path", "")
        match = re.search(r"/servers/(?P<server_id>\d+)/mcp", path)
        
        if match:
            server_id = match.group("server_id")
            server_id_var.set(server_id)  # ✅ Already implemented
            
        # Continue with request processing...

Phase 3: Error Handling & Validation

# Add proper error handling for virtual server validation
async def validate_server_access(server_id: str, resource_type: str, resource_name: str):
    """Validate that a resource belongs to the specified virtual server."""
    async with get_db() as db:
        server = await server_service.get_server(db, server_id)
        if not server:
            raise HTTPException(404, f"Virtual server {server_id} not found")
        
        # Check if resource is associated with this server
        # Implementation varies by resource type (tool/resource/prompt)

---

📊 Implementation Components

Component	Status	Required Changes
streamablehttp_transport.py	❌ Missing	Add list_resources(), list_prompts(), update call_tool()
Context Variable Handling	✅ Partial	Already extracts server_id, needs consistent usage
Error Handling	❌ Missing	Add virtual server validation and proper error responses
Path Routing	✅ Working	Already routes /servers/{id}/mcp correctly
Authentication	✅ Working	Already handles auth in middleware

---

🔄 Roll-out Plan

1. Phase 0: Analysis and testing setup

   • Create integration tests comparing SSE vs Streamable HTTP behavior
   • Document current behavior differences

2. Phase 1: Core method implementations

   • Add missing list_resources() and list_prompts() handlers
   • Update call_tool() with server context validation
   • Add get_prompt() and read_resource() handlers

3. Phase 2: Error handling and edge cases

   • Proper validation for virtual server resource access
   • Consistent error responses matching SSE transport
   • Fallback behavior for malformed requests

4. Phase 3: Testing and validation

   • Comprehensive integration tests
   • Performance comparison with SSE transport
   • Documentation updates

---

🧪 Testing Strategy

Unit Tests

def test_streamable_http_virtual_server_tools():
    """Test that tools/list respects virtual server scoping."""
    # Connect to /servers/1/mcp
    # Verify only server 1 tools are returned
    
def test_streamable_http_virtual_server_resources():
    """Test that resources/list respects virtual server scoping."""
    # Connect to /servers/1/mcp  
    # Verify only server 1 resources are returned

def test_streamable_http_global_fallback():
    """Test that /mcp (no server) returns global catalog."""
    # Connect to /mcp (global endpoint)
    # Verify all tools/resources/prompts are returned

Integration Tests

def test_transport_parity():
    """Test that SSE and Streamable HTTP return identical results."""
    # Connect to same virtual server via both transports
    # Compare tools/list, resources/list, prompts/list responses
    # Verify identical scoping behavior

---

📝 Dependencies & Prerequisites

Technical Dependencies:

• Existing tool_service.list_server_tools() methods (✅ implemented)
• Existing resource_service.list_server_resources() methods (✅ implemented)
• Existing prompt_service.list_server_prompts() methods (✅ implemented)
• Context variable infrastructure (✅ partially implemented)

Related Issues:

• May interact with virtual server authentication/authorization
• Should align with any upcoming RBAC features ([SECURITY FEATURE]: Role-Based Access Control (RBAC) - User/Team/Global Scopes for full multi-tenancy support #283)
• May need updates for plugin framework integration

Timeline:

• Target Release: v0.4.0 or v0.5.0 (bugfix/enhancement)
• Priority: High (transport consistency issue)
• Effort: Medium (extend existing patterns)

---

📋 Success Metrics

Functional:

• ✅ tools/list via Streamable HTTP respects virtual server scoping
• ✅ resources/list via Streamable HTTP respects virtual server scoping
• ✅ prompts/list via Streamable HTTP respects virtual server scoping
• ✅ All MCP operations work identically between SSE and Streamable HTTP

Technical:

• ✅ No performance regression in Streamable HTTP transport
• ✅ Maintains backward compatibility for global /mcp endpoint
• ✅ Proper error handling for invalid virtual server requests
• ✅ 100% test coverage for virtual server scoping logic

This enhancement ensures transport consistency and maintains the virtual server security model across all supported MCP transports.

[kumar-tiger]

Tools are working with streamable http by changing the
match = re.search(r"/servers/(?P<server_id>\d+)/mcp", path)
with following in streamablehttp_transport.py
match = re.search(r"/servers/(?P<server_id>\w+)/mcp", path)