Add MCP Resources Tracking Support

[naji247]

Overview

Add tracking support for MCP resource operations to match the existing tool tracking capabilities. Resources provide contextual data to language models (files, schemas, application data) and we should track how clients discover and access these resources.

Current State

MCPcat already defines the event types for resources but lacks the implementation:

• MCP_RESOURCES_LIST = "mcp:resources/list"
• MCP_RESOURCES_READ = "mcp:resources/read"
• MCP_RESOURCES_SUBSCRIBE = "mcp:resources/subscribe"
• MCP_RESOURCES_UNSUBSCRIBE = "mcp:resources/unsubscribe"
• MCP_RESOURCES_TEMPLATES_LIST = "mcp:resources/templates/list"

Implementation Requirements

Track resource operations with the same scope and pattern as tool tracking:

1. List Resources (resources/list)

Track when clients request the list of available resources:

async def wrapped_list_resources_handler(request: ListResourcesRequest) -> ServerResult:
    session_id = get_server_session_id(server)
    request_context = safe_request_context(server)
    get_client_info_from_request_context(server, request_context)
    identify_session(server, request, request_context)
    
    event = UnredactedEvent(
        session_id=session_id,
        timestamp=datetime.now(timezone.utc),
        parameters=request.params.model_dump() if request.params else {},
        event_type=EventType.MCP_RESOURCES_LIST.value,
    )
    
    result = await original_list_resources_handler(request)
    event.response = result.model_dump() if result else None
    event_queue.publish_event(server, event)
    return result

2. Read Resource (resources/read)

Track each resource read operation:

async def wrapped_read_resource_handler(request: ReadResourceRequest) -> ServerResult:
    resource_uri = request.params.uri
    session_id = get_server_session_id(server)
    request_context = safe_request_context(server)
    get_client_info_from_request_context(server, request_context)
    identify_session(server, request, request_context)
    
    event = UnredactedEvent(
        session_id=session_id,
        timestamp=datetime.now(timezone.utc),
        parameters=request.params.model_dump() if request.params else {},
        event_type=EventType.MCP_RESOURCES_READ.value,
        resource_name=resource_uri,  # Store URI in resource_name field
    )
    
    if data.options.enable_tracing:
        try:
            result = await original_read_resource_handler(request)
            is_error, error_message = is_mcp_error_response(result)
            event.is_error = is_error
            event.error = {"message": error_message} if is_error else None
            event.response = result.model_dump() if result else None
            event_queue.publish_event(server, event)
            return result
        except Exception as e:
            event.is_error = True
            event.error = {"message": str(e)}
            event_queue.publish_event(server, event)
            raise
    else:
        return await original_read_resource_handler(request)

3. Subscribe/Unsubscribe (resources/subscribe, resources/unsubscribe)

Track resource subscription lifecycle:

async def wrapped_subscribe_handler(request: SubscribeRequest) -> ServerResult:
    resource_uri = request.params.uri
    session_id = get_server_session_id(server)
    request_context = safe_request_context(server)
    get_client_info_from_request_context(server, request_context)
    identify_session(server, request, request_context)
    
    event = UnredactedEvent(
        session_id=session_id,
        timestamp=datetime.now(timezone.utc),
        parameters=request.params.model_dump() if request.params else {},
        event_type=EventType.MCP_RESOURCES_SUBSCRIBE.value,
        resource_name=resource_uri,
    )
    
    result = await original_subscribe_handler(request)
    event.response = result.model_dump() if result else None
    event_queue.publish_event(server, event)
    return result

4. List Resource Templates (resources/templates/list)

Track template listing requests:

async def wrapped_list_templates_handler(request: ListResourceTemplatesRequest) -> ServerResult:
    # Similar pattern to list_resources_handler
    # Track with EventType.MCP_RESOURCES_TEMPLATES_LIST

Key Design Decisions

1. Consistent with Tool Tracking: Use the same patterns and field mappings:

   • resource_name field stores the resource URI (like tool names)
   • parameters stores request parameters
   • response stores the full response
   • Error tracking with is_error and error fields

2. Respect enable_tracing Option: Only track detailed responses when tracing is enabled, matching tool behavior

3. No Response Modification: Unlike tools (which inject context parameters), resource responses are passed through unmodified

4. No MCPcat Resources: Unlike tools (which inject get_more_tools), we don't add special MCPcat resources

Implementation Location

Add handlers in src/mcpcat/modules/overrides/mcp_server.py alongside existing tool handlers:

def override_lowlevel_mcp_server(server: Server, data: MCPCatData) -> None:
    # ... existing tool handler setup ...
    
    # Store original resource handlers
    original_list_resources_handler = server.request_handlers.get(ListResourcesRequest)
    original_read_resource_handler = server.request_handlers.get(ReadResourceRequest)
    original_subscribe_handler = server.request_handlers.get(SubscribeRequest)
    original_unsubscribe_handler = server.request_handlers.get(UnsubscribeRequest)
    original_list_templates_handler = server.request_handlers.get(ListResourceTemplatesRequest)
    
    # ... wrapper handler definitions ...
    
    # Register wrapped handlers
    server.request_handlers[ListResourcesRequest] = wrapped_list_resources_handler
    server.request_handlers[ReadResourceRequest] = wrapped_read_resource_handler
    server.request_handlers[SubscribeRequest] = wrapped_subscribe_handler
    server.request_handlers[UnsubscribeRequest] = wrapped_unsubscribe_handler
    server.request_handlers[ListResourceTemplatesRequest] = wrapped_list_templates_handler

Acceptance Criteria

• Resource list requests are tracked with full response
• Resource read operations are tracked with URI and content
• Subscribe/unsubscribe events are captured with resource URIs
• Template listing is tracked
• Error cases are properly tracked with error details
• Tracking respects enable_tracing option for detailed responses
• Implementation follows the same patterns as tool tracking
• No modification of resource responses
• Events are properly sent to MCPcat API or telemetry exporters

Testing Considerations

• Test with various resource URI schemes (file://, https://, custom)
• Verify pagination handling in list requests
• Test error scenarios (missing resources, invalid URIs)
• Ensure subscription lifecycle is properly tracked
• Verify that resource responses are not modified