Add coverage for remaining branch gaps

- Add pragma for defensive _meta checks in session.py (unreachable code)
- Add tests for router loop continuation (non-matching routers)
- Add tests for enable_tasks with custom store/queue
- Add tests for skipping default handlers when custom registered

Author: maxisbey

src/mcp/server/session.py

@@ -536,7 +536,9 @@ def _build_elicit_request(
 
         # Add related-task metadata if in task mode
         if task_id is not None:
-            if "_meta" not in params_data:
+            # Defensive check: _meta can't exist currently since ElicitRequestFormParams
+            # doesn't pass meta to model_dump, but guard against future changes.
+            if "_meta" not in params_data:  # pragma: no cover
                 params_data["_meta"] = {}
             params_data["_meta"]["io.modelcontextprotocol/related-task"] = {"taskId": task_id}
 
@@ -594,7 +596,9 @@ def _build_create_message_request(
 
         # Add related-task metadata if in task mode
         if task_id is not None:
-            if "_meta" not in params_data:
+            # Defensive check: _meta can't exist currently since CreateMessageRequestParams
+            # doesn't pass meta to model_dump, but guard against future changes.
+            if "_meta" not in params_data:  # pragma: no cover
                 params_data["_meta"] = {}
             params_data["_meta"]["io.modelcontextprotocol/related-task"] = {"taskId": task_id}
 

tests/experimental/tasks/server/test_run_task_flow.py

@@ -9,6 +9,7 @@
 These are integration tests that verify the complete flow works end-to-end.
 """
 
+from datetime import datetime, timezone
 from typing import Any
 
 import anyio
@@ -214,6 +215,92 @@ async def test_enable_tasks_auto_registers_handlers() -> None:
     assert caps_after.tasks.cancel is not None
 
 
+@pytest.mark.anyio
+async def test_enable_tasks_with_custom_store_and_queue() -> None:
+    """Test that enable_tasks() uses provided store and queue instead of defaults."""
+    from mcp.shared.experimental.tasks.in_memory_task_store import InMemoryTaskStore
+    from mcp.shared.experimental.tasks.message_queue import InMemoryTaskMessageQueue
+
+    server = Server("test-custom-store-queue")
+
+    # Create custom store and queue
+    custom_store = InMemoryTaskStore()
+    custom_queue = InMemoryTaskMessageQueue()
+
+    # Enable tasks with custom implementations
+    task_support = server.experimental.enable_tasks(store=custom_store, queue=custom_queue)
+
+    # Verify our custom implementations are used
+    assert task_support.store is custom_store
+    assert task_support.queue is custom_queue
+
+
+@pytest.mark.anyio
+async def test_enable_tasks_skips_default_handlers_when_custom_registered() -> None:
+    """Test that enable_tasks() doesn't override already-registered handlers."""
+    from mcp.types import (
+        CancelTaskRequest,
+        CancelTaskResult,
+        GetTaskPayloadRequest,
+        GetTaskPayloadResult,
+        GetTaskRequest,
+        GetTaskResult,
+        ListTasksRequest,
+        ListTasksResult,
+    )
+
+    server = Server("test-custom-handlers")
+
+    # Track which custom handlers were called
+    custom_handlers_called: list[str] = []
+
+    # Use a fixed timestamp for deterministic tests
+    fixed_time = datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
+
+    # Register custom handlers BEFORE enable_tasks
+    @server.experimental.get_task()
+    async def custom_get_task(req: GetTaskRequest) -> GetTaskResult:
+        custom_handlers_called.append("get_task")
+        return GetTaskResult(
+            taskId="custom",
+            status="working",
+            createdAt=fixed_time,
+            lastUpdatedAt=fixed_time,
+            ttl=60000,
+        )
+
+    @server.experimental.get_task_result()
+    async def custom_get_task_result(req: GetTaskPayloadRequest) -> GetTaskPayloadResult:
+        custom_handlers_called.append("get_task_result")
+        return GetTaskPayloadResult()
+
+    @server.experimental.list_tasks()
+    async def custom_list_tasks(req: ListTasksRequest) -> ListTasksResult:
+        custom_handlers_called.append("list_tasks")
+        return ListTasksResult(tasks=[])
+
+    @server.experimental.cancel_task()
+    async def custom_cancel_task(req: CancelTaskRequest) -> CancelTaskResult:
+        custom_handlers_called.append("cancel_task")
+        return CancelTaskResult(
+            taskId="custom",
+            status="cancelled",
+            createdAt=fixed_time,
+            lastUpdatedAt=fixed_time,
+            ttl=60000,
+        )
+
+    # Now enable tasks - should NOT override our custom handlers
+    server.experimental.enable_tasks()
+
+    # Verify our custom handlers are still registered (not replaced by defaults)
+    # The handlers dict should contain our custom handlers
+    assert GetTaskRequest in server.request_handlers
+    assert GetTaskPayloadRequest in server.request_handlers
+    assert ListTasksRequest in server.request_handlers
+    assert CancelTaskRequest in server.request_handlers
+
+
 @pytest.mark.anyio
 async def test_run_task_without_enable_tasks_raises() -> None:
     """Test that run_task raises when enable_tasks() wasn't called."""

tests/experimental/tasks/server/test_server.py

@@ -864,3 +864,129 @@ def route_error(self, request_id: str | int, error: ErrorData) -> bool:
         await server_to_client_receive.aclose()
         await client_to_server_send.aclose()
         await client_to_server_receive.aclose()
+
+
+@pytest.mark.anyio
+async def test_response_routing_skips_non_matching_routers() -> None:
+    """Test that routing continues to next router when first doesn't match."""
+    from mcp.shared.session import ResponseRouter
+
+    server_to_client_send, server_to_client_receive = anyio.create_memory_object_stream[SessionMessage](10)
+    client_to_server_send, client_to_server_receive = anyio.create_memory_object_stream[SessionMessage](10)
+
+    # Track which routers were called
+    router_calls: list[str] = []
+    response_received = anyio.Event()
+
+    class NonMatchingRouter(ResponseRouter):
+        def route_response(self, request_id: str | int, response: dict[str, Any]) -> bool:
+            router_calls.append("non_matching_response")
+            return False  # Doesn't handle it
+
+        def route_error(self, request_id: str | int, error: ErrorData) -> bool:
+            router_calls.append("non_matching_error")
+            return False  # Doesn't handle it
+
+    class MatchingRouter(ResponseRouter):
+        def route_response(self, request_id: str | int, response: dict[str, Any]) -> bool:
+            router_calls.append("matching_response")
+            response_received.set()
+            return True  # Handles it
+
+        def route_error(self, request_id: str | int, error: ErrorData) -> bool:
+            router_calls.append("matching_error")
+            response_received.set()
+            return True  # Handles it
+
+    try:
+        async with ServerSession(
+            client_to_server_receive,
+            server_to_client_send,
+            InitializationOptions(
+                server_name="test-server",
+                server_version="1.0.0",
+                capabilities=ServerCapabilities(),
+            ),
+        ) as server_session:
+            # Add non-matching router first, then matching router
+            server_session.add_response_router(NonMatchingRouter())
+            server_session.add_response_router(MatchingRouter())
+
+            # Send a response - should skip first router and be handled by second
+            response = JSONRPCResponse(jsonrpc="2.0", id="test-req-1", result={"status": "ok"})
+            message = SessionMessage(message=JSONRPCMessage(response))
+            await client_to_server_send.send(message)
+
+            with anyio.fail_after(5):
+                await response_received.wait()
+
+            # Verify both routers were called (first returned False, second returned True)
+            assert router_calls == ["non_matching_response", "matching_response"]
+    finally:
+        await server_to_client_send.aclose()
+        await server_to_client_receive.aclose()
+        await client_to_server_send.aclose()
+        await client_to_server_receive.aclose()
+
+
+@pytest.mark.anyio
+async def test_error_routing_skips_non_matching_routers() -> None:
+    """Test that error routing continues to next router when first doesn't match."""
+    from mcp.shared.session import ResponseRouter
+
+    server_to_client_send, server_to_client_receive = anyio.create_memory_object_stream[SessionMessage](10)
+    client_to_server_send, client_to_server_receive = anyio.create_memory_object_stream[SessionMessage](10)
+
+    # Track which routers were called
+    router_calls: list[str] = []
+    error_received = anyio.Event()
+
+    class NonMatchingRouter(ResponseRouter):
+        def route_response(self, request_id: str | int, response: dict[str, Any]) -> bool:
+            router_calls.append("non_matching_response")
+            return False
+
+        def route_error(self, request_id: str | int, error: ErrorData) -> bool:
+            router_calls.append("non_matching_error")
+            return False  # Doesn't handle it
+
+    class MatchingRouter(ResponseRouter):
+        def route_response(self, request_id: str | int, response: dict[str, Any]) -> bool:
+            router_calls.append("matching_response")
+            return True
+
+        def route_error(self, request_id: str | int, error: ErrorData) -> bool:
+            router_calls.append("matching_error")
+            error_received.set()
+            return True  # Handles it
+
+    try:
+        async with ServerSession(
+            client_to_server_receive,
+            server_to_client_send,
+            InitializationOptions(
+                server_name="test-server",
+                server_version="1.0.0",
+                capabilities=ServerCapabilities(),
+            ),
+        ) as server_session:
+            # Add non-matching router first, then matching router
+            server_session.add_response_router(NonMatchingRouter())
+            server_session.add_response_router(MatchingRouter())
+
+            # Send an error - should skip first router and be handled by second
+            error_data = ErrorData(code=-32600, message="Test error")
+            error_response = JSONRPCError(jsonrpc="2.0", id="test-req-2", error=error_data)
+            message = SessionMessage(message=JSONRPCMessage(error_response))
+            await client_to_server_send.send(message)
+
+            with anyio.fail_after(5):
+                await error_received.wait()
+
+            # Verify both routers were called (first returned False, second returned True)
+            assert router_calls == ["non_matching_error", "matching_error"]
+    finally:
+        await server_to_client_send.aclose()
+        await server_to_client_receive.aclose()
+        await client_to_server_send.aclose()
+        await client_to_server_receive.aclose()