[minimcp] Improve test coverage for edge cases and error handling

- Add tests for error scenarios across all managers and transports
- Add edge case coverage for MiniMCP core orchestrator
- Add validation and error handling tests
- Clean up test imports

Author: sreenaths

tests/server/minimcp/unit/managers/test_prompt_manager.py

@@ -8,6 +8,7 @@
 from mcp.server.lowlevel.server import Server
 from mcp.server.minimcp.exceptions import InvalidArgumentsError, MCPFuncError, MCPRuntimeError, PrimitiveError
 from mcp.server.minimcp.managers.prompt_manager import PromptDefinition, PromptManager
+from mcp.server.minimcp.utils.mcp_func import MCPFunc
 
 pytestmark = pytest.mark.anyio
 
@@ -916,3 +917,29 @@ def no_args_prompt() -> str:
         # Should be callable with empty args
         get_result = await prompt_manager.get("no_args_prompt", None)
         assert len(get_result.messages) == 1
+
+    def test_get_arguments_without_properties(self, prompt_manager: PromptManager):
+        """Test _get_arguments when input_schema has no properties.
+
+        A function with no parameters will naturally have an input_schema
+        without a 'properties' key, which tests the branch where
+        'properties' is not in input_schema.
+        """
+
+        def simple_func() -> str:
+            """A simple function with no parameters"""
+            return "result"
+
+        mcp_func = MCPFunc(simple_func)
+
+        # Verify the schema doesn't have properties (or has empty properties)
+        # This tests the code path where "properties" not in input_schema
+        arguments = prompt_manager._get_arguments(mcp_func)
+        assert arguments == []
+
+    async def test_validate_args_with_none(self, prompt_manager: PromptManager):
+        """Test _validate_args when prompt_arguments is None."""
+        # This should return early without raising an error
+        prompt_manager._validate_args(None, {"some": "args"})
+        prompt_manager._validate_args(None, None)
+        # If we get here without exception, the test passes

tests/server/minimcp/unit/managers/test_tool_manager.py

@@ -1,13 +1,20 @@
 from typing import Any
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 import anyio
 import pytest
 from pydantic import BaseModel, Field
 
 import mcp.types as types
+from mcp.server.fastmcp.utilities.func_metadata import FuncMetadata
 from mcp.server.lowlevel.server import Server
-from mcp.server.minimcp.exceptions import InvalidArgumentsError, MCPFuncError, MCPRuntimeError, PrimitiveError
+from mcp.server.minimcp.exceptions import (
+    InvalidArgumentsError,
+    MCPFuncError,
+    MCPRuntimeError,
+    PrimitiveError,
+    ToolMCPRuntimeError,
+)
 from mcp.server.minimcp.managers.tool_manager import ToolDefinition, ToolManager
 
 pytestmark = pytest.mark.anyio
@@ -827,3 +834,31 @@ def error_tool(trigger: bool) -> str:
         tools = tool_manager.list()
         assert len(tools) == 1
         assert tools[0].name == "error_tool"
+
+    async def test_tool_convert_result_exception(self, tool_manager: ToolManager):
+        """Test that exceptions during result conversion are handled properly."""
+
+        def simple_tool(value: str) -> str:
+            """A simple tool"""
+            return value
+
+        tool_manager.add(simple_tool)
+
+        # Patch the convert_result method on FuncMetadata class
+        with patch.object(FuncMetadata, "convert_result", side_effect=ValueError("Conversion failed")):
+            with pytest.raises(ToolMCPRuntimeError, match="Error calling tool simple_tool"):
+                await tool_manager._call("simple_tool", {"value": "test"})
+
+    async def test_tool_call_wrapper_mcp_runtime_error(self, tool_manager: ToolManager):
+        """Test that call() wrapper properly converts ToolMCPRuntimeError to MCPRuntimeError."""
+
+        def simple_tool(value: str) -> str:
+            """A simple tool"""
+            return value
+
+        tool_manager.add(simple_tool)
+
+        # Patch the convert_result method on FuncMetadata class
+        with patch.object(FuncMetadata, "convert_result", side_effect=RuntimeError("Runtime error")):
+            with pytest.raises(MCPRuntimeError, match="Error calling tool simple_tool"):
+                await tool_manager.call("simple_tool", {"value": "test"})

tests/server/minimcp/unit/test_minimcp.py

@@ -1,5 +1,6 @@
 import json
 from collections.abc import Coroutine
+from datetime import datetime
 from typing import Any
 from unittest.mock import AsyncMock, Mock, patch
 
@@ -10,10 +11,15 @@
 from mcp.server.lowlevel.server import NotificationOptions, Server
 from mcp.server.minimcp.exceptions import (
     ContextError,
+    InternalMCPError,
+    InvalidArgumentsError,
     InvalidJSONError,
     InvalidJSONRPCMessageError,
     InvalidMessageError,
+    MCPRuntimeError,
+    PrimitiveError,
     RequestHandlerNotFoundError,
+    ResourceNotFoundError,
     UnsupportedMessageTypeError,
 )
 from mcp.server.minimcp.managers.context_manager import Context, ContextManager
@@ -711,13 +717,10 @@ async def test_error_metadata_in_response(self):
         assert "isoTimestamp" in error_data
 
         # Verify timestamp is valid ISO format
-        from datetime import datetime
-
         datetime.fromisoformat(error_data["isoTimestamp"])
 
     async def test_process_error_with_internal_mcp_error(self):
         """Test _process_error method with InternalMCPError that has data."""
-        from mcp.server.minimcp.exceptions import ResourceNotFoundError
 
         server: MiniMCP[Any] = MiniMCP(name="process-error-test")
 
@@ -736,11 +739,6 @@ async def test_process_error_with_internal_mcp_error(self):
 
     async def test_handle_with_different_error_types(self):
         """Test handling different types of MiniMCP errors."""
-        from mcp.server.minimcp.exceptions import (
-            InvalidArgumentsError,
-            MCPRuntimeError,
-            PrimitiveError,
-        )
 
         server: MiniMCP[Any] = MiniMCP(name="error-types-test")
 
@@ -838,3 +836,91 @@ async def test_limiter_integration_with_errors(self):
         for i, response in enumerate(results):
             assert response["id"] == i
             assert "error" in response
+
+    async def test_unsupported_message_type_error(self):
+        """Test handling of UnsupportedMessageTypeError."""
+        server: MiniMCP[Any] = MiniMCP(name="test-server")
+
+        # Create a message that will trigger UnsupportedMessageTypeError
+        # This happens when the message is valid JSON-RPC but not a request or notification
+        with patch("mcp.server.minimcp.minimcp.MiniMCP._handle_rpc_msg") as mock_handle:
+            from mcp.server.minimcp.exceptions import UnsupportedMessageTypeError
+
+            mock_handle.side_effect = UnsupportedMessageTypeError("Unsupported message type")
+
+            message = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "test", "params": {}})
+            result = await server.handle(message)
+
+            # Should return an error response
+            response = json.loads(result)  # type: ignore
+            assert "error" in response
+            assert response["id"] == 1
+
+    async def test_request_handler_not_found_error(self):
+        """Test handling of RequestHandlerNotFoundError."""
+        server: MiniMCP[Any] = MiniMCP(name="test-server")
+
+        # Create a message with a method that doesn't exist in MiniMCP
+        message = json.dumps(
+            {"jsonrpc": "2.0", "id": 1, "method": "resources/subscribe", "params": {"uri": "file:///config.json"}}
+        )
+        result = await server.handle(message)
+
+        # Should return a METHOD_NOT_FOUND error
+        response = json.loads(result)  # type: ignore
+        assert "error" in response
+        assert response["error"]["code"] == types.METHOD_NOT_FOUND
+        assert response["id"] == 1
+
+    async def test_resource_not_found_error(self):
+        """Test handling of ResourceNotFoundError."""
+        server: MiniMCP[Any] = MiniMCP(name="test-server")
+
+        # Try to read a resource that doesn't exist
+        message = json.dumps(
+            {"jsonrpc": "2.0", "id": 1, "method": "resources/read", "params": {"uri": "nonexistent://resource"}}
+        )
+        result = await server.handle(message)
+
+        # Should return a RESOURCE_NOT_FOUND error
+        response = json.loads(result)  # type: ignore
+        assert "error" in response
+        assert response["error"]["code"] == types.RESOURCE_NOT_FOUND
+        assert response["id"] == 1
+
+    async def test_internal_mcp_error(self):
+        """Test handling of InternalMCPError."""
+
+        server: MiniMCP[Any] = MiniMCP(name="test-server")
+
+        # Mock _handle_rpc_msg to raise InternalMCPError
+        with patch("mcp.server.minimcp.minimcp.MiniMCP._handle_rpc_msg") as mock_handle:
+            mock_handle.side_effect = InternalMCPError("Internal error")
+
+            message = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "test", "params": {}})
+            result = await server.handle(message)
+
+            # Should return an INTERNAL_ERROR response
+            response = json.loads(result)  # type: ignore
+            assert "error" in response
+            assert response["error"]["code"] == types.INTERNAL_ERROR
+            assert response["id"] == 1
+
+    async def test_error_type_checking_branches(self):
+        """Test the error type checking branches in _parse_message."""
+        server: MiniMCP[Any] = MiniMCP(name="test-server")
+
+        # Test with invalid JSON that triggers json_invalid error
+        invalid_json = "{invalid json"
+        with pytest.raises(InvalidMessageError):
+            await server.handle(invalid_json)
+
+        # Test with valid JSON but invalid JSON-RPC (missing jsonrpc field)
+        invalid_jsonrpc = json.dumps({"id": 1, "method": "test"})
+        with pytest.raises(InvalidMessageError):
+            await server.handle(invalid_jsonrpc)
+
+        # Test with wrong jsonrpc version
+        wrong_version = json.dumps({"jsonrpc": "1.0", "id": 1, "method": "test", "params": {}})
+        with pytest.raises(InvalidMessageError):
+            await server.handle(wrong_version)

tests/server/minimcp/unit/transports/test_base_http_transport.py

@@ -2,7 +2,7 @@
 from collections.abc import Mapping
 from http import HTTPStatus
 from typing import Any, NamedTuple
-from unittest.mock import ANY, AsyncMock
+from unittest.mock import ANY, AsyncMock, patch
 
 import anyio
 import pytest
@@ -11,6 +11,7 @@
 from starlette.responses import Response
 from typing_extensions import override
 
+from mcp.server.minimcp.exceptions import InvalidMessageError
 from mcp.server.minimcp.minimcp import MiniMCP
 from mcp.server.minimcp.transports.base_http import MEDIA_TYPE_JSON, BaseHTTPTransport, RequestValidationError
 from mcp.shared.version import LATEST_PROTOCOL_VERSION
@@ -126,6 +127,18 @@ class TestBaseHTTPTransportHeaderValidation:
     def transport(self) -> BaseHTTPTransport[Any]:
         return MockHTTPTransport(AsyncMock(spec=MiniMCP[Any]))
 
+    @pytest.fixture
+    def accept_content_types(self) -> str:
+        return "application/json"
+
+    @pytest.fixture
+    def request_headers(self, accept_content_types: str) -> dict[str, str]:
+        return {
+            "Content-Type": "application/json",
+            "Accept": accept_content_types,
+            "MCP-Protocol-Version": LATEST_PROTOCOL_VERSION,
+        }
+
     def test_validate_accept_headers_valid(self, transport: BaseHTTPTransport[Any]):
         """Test validate accept headers with valid headers."""
         headers = {"Accept": "application/json, text/plain"}
@@ -279,3 +292,20 @@ def test_whitespace_in_headers(self, transport: BaseHTTPTransport[Any]):
 
         # Content-Type header test
         transport._validate_content_type(headers)
+
+    async def test_handle_post_request_with_invalid_message_error(self, request_headers: dict[str, str]):
+        """Test _handle_post_request when InvalidMessageError is raised."""
+
+        server = MiniMCP[Any](name="test-server", version="1.0.0")
+        transport = MockHTTPTransport(server)
+
+        # Create an invalid message that will trigger InvalidMessageError
+        invalid_message = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "test"})
+
+        # Mock handle to raise InvalidMessageError with a response
+        error_response = '{"jsonrpc": "2.0", "error": {"code": -32600, "message": "Invalid Request"}, "id": 1}'
+        with patch.object(server, "handle", side_effect=InvalidMessageError("Invalid", error_response)):
+            result = await transport._handle_post_request(request_headers, invalid_message, None)
+
+            assert result.status_code == HTTPStatus.BAD_REQUEST
+            assert result.content == error_response

tests/server/minimcp/unit/transports/test_http_transport.py

@@ -5,6 +5,7 @@
 
 import anyio
 import pytest
+from starlette.requests import Request
 
 from mcp.server.minimcp.minimcp import MiniMCP
 from mcp.server.minimcp.transports.base_http import MEDIA_TYPE_JSON, RequestValidationError
@@ -374,6 +375,18 @@ class TestHTTPTransportHeaderValidation:
     def transport(self) -> HTTPTransport[Any]:
         return HTTPTransport[Any](AsyncMock(spec=MiniMCP[Any]))
 
+    @pytest.fixture
+    def accept_content_types(self) -> str:
+        return "application/json"
+
+    @pytest.fixture
+    def request_headers(self, accept_content_types: str) -> dict[str, str]:
+        return {
+            "Content-Type": "application/json",
+            "Accept": accept_content_types,
+            "MCP-Protocol-Version": LATEST_PROTOCOL_VERSION,
+        }
+
     def test_validate_accept_headers_valid(self, transport: HTTPTransport[Any]):
         """Test validate accept headers with valid headers."""
         headers = {"Accept": "application/json, text/plain"}
@@ -527,3 +540,48 @@ def test_whitespace_in_headers(self, transport: HTTPTransport[Any]):
 
         # Content-Type header test
         transport._validate_content_type(headers)
+
+    async def test_starlette_dispatch(self, request_headers: dict[str, str]):
+        """Test starlette_dispatch method."""
+
+        server = MiniMCP[Any](name="test-server", version="1.0.0")
+        transport = HTTPTransport(server)
+
+        # Create a mock request
+        init_message = json.dumps(
+            {
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "initialize",
+                "params": {
+                    "protocolVersion": LATEST_PROTOCOL_VERSION,
+                    "capabilities": {},
+                    "clientInfo": {"name": "test", "version": "1.0"},
+                },
+            }
+        )
+
+        # Mock Starlette request
+        scope = {
+            "type": "http",
+            "method": "POST",
+            "headers": [(k.lower().encode(), v.encode()) for k, v in request_headers.items()],
+        }
+        request = Request(scope)
+        request._body = init_message.encode()
+
+        response = await transport.starlette_dispatch(request)
+
+        assert response.status_code == HTTPStatus.OK
+        assert response.media_type == MEDIA_TYPE_JSON
+
+    async def test_as_starlette(self, request_headers: dict[str, str]):
+        """Test as_starlette method."""
+        server = MiniMCP[Any](name="test-server", version="1.0.0")
+        transport = HTTPTransport(server)
+
+        app = transport.as_starlette(path="/mcp", debug=True)
+
+        # Verify app is created
+        assert app is not None
+        assert len(app.routes) == 1