fix: Resolve test failures in HTTP MCP support tests

1. Fragment filtering test: Added proper assertion for MCPToolWrapper call
   and verified the mock tool is returned in the result list.

2. Logger patching tests: Changed from patching instance method to patching
   class method to avoid Pydantic's __delattr__ restriction. This prevents
   AttributeError when the patch context manager tries to restore the
   original attribute.

Co-Authored-By: João <joao@crewai.com>

lib/crewai/src/crewai/agent/core.py

@@ -713,7 +713,7 @@ class Agent(BaseAgent):
         """Get tools from legacy string-based MCP references.
 
         This method maintains backwards compatibility with string-based
-        MCP references (https://... and crewai-amp:...).
+        MCP references (http://..., https://..., and crewai-amp:...).
 
         Args:
             mcp_ref: String reference to MCP server.
@@ -723,12 +723,12 @@ class Agent(BaseAgent):
         """
         if mcp_ref.startswith("crewai-amp:"):
             return self._get_amp_mcp_tools(mcp_ref)
-        if mcp_ref.startswith("https://"):
+        if mcp_ref.startswith(("http://", "https://")):
             return self._get_external_mcp_tools(mcp_ref)
         return []
 
     def _get_external_mcp_tools(self, mcp_ref: str) -> list[BaseTool]:
-        """Get tools from external HTTPS MCP server with graceful error handling."""
+        """Get tools from external HTTP/HTTPS MCP server with graceful error handling."""
         from crewai.tools.mcp_tool_wrapper import MCPToolWrapper
 
         # Parse server URL and optional tool name
@@ -737,6 +737,15 @@ class Agent(BaseAgent):
         else:
             server_url, specific_tool = mcp_ref, None
 
+        parsed_url = urlparse(server_url)
+        if parsed_url.scheme == "http":
+            self._logger.log(
+                "warning",
+                f"Using http:// for MCP server '{server_url}'. "
+                "This is intended for local development only. "
+                "Use https:// in production to ensure secure communication.",
+            )
+
         server_params = {"url": server_url}
         server_name = self._extract_server_name(server_url)
 

lib/crewai/src/crewai/agents/agent_builder/base_agent.py

@@ -197,7 +197,7 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
     )
     mcps: list[str | MCPServerConfig] | None = Field(
         default=None,
-        description="List of MCP server references. Supports 'https://server.com/path' for external servers and 'crewai-amp:mcp-name' for AMP marketplace. Use '#tool_name' suffix for specific tools.",
+        description="List of MCP server references. Supports 'http://localhost:port/path' or 'https://server.com/path' for external servers and 'crewai-amp:mcp-name' for AMP marketplace. Use '#tool_name' suffix for specific tools. Note: http:// is intended for local development only; use https:// in production.",
     )
 
     @model_validator(mode="before")
@@ -268,12 +268,12 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
         validated_mcps = []
         for mcp in mcps:
             if isinstance(mcp, str):
-                if mcp.startswith(("https://", "crewai-amp:")):
+                if mcp.startswith(("http://", "https://", "crewai-amp:")):
                     validated_mcps.append(mcp)
                 else:
                     raise ValueError(
                         f"Invalid MCP reference: {mcp}. "
-                        "String references must start with 'https://' or 'crewai-amp:'"
+                        "String references must start with 'http://', 'https://', or 'crewai-amp:'"
                     )
 
             elif isinstance(mcp, (MCPServerConfig)):

lib/crewai/src/crewai/mcp/config.py

@@ -63,6 +63,11 @@ class MCPServerHTTP(BaseModel):
             headers={"Authorization": "Bearer ..."},
             cache_tools_list=True,
         )
+
+        mcp_server = MCPServerHTTP(
+            url="http://localhost:8000/mcp",
+            cache_tools_list=True,
+        )
         ```
     """
 

lib/crewai/src/crewai/mcp/transports/http.py

@@ -28,6 +28,11 @@ class HTTPTransport(BaseTransport):
             url="https://api.example.com/mcp",
             headers={"Authorization": "Bearer ..."}
         )
+
+        transport = HTTPTransport(
+            url="http://localhost:8000/mcp"
+        )
+
         async with transport:
             # Use transport...
         ```

lib/crewai/tests/mcp/test_http_mcp_support.py

@@ -0,0 +1,216 @@
+"""Tests for HTTP MCP server support (issue #3876)."""
+
+import pytest
+from unittest.mock import MagicMock, patch
+
+from crewai.agent.core import Agent
+from crewai.mcp.config import MCPServerHTTP
+
+
+class TestHTTPMCPValidation:
+    """Test validation of HTTP MCP URLs."""
+
+    def test_validator_accepts_http_urls(self):
+        """Test that validator accepts http:// URLs."""
+        agent = Agent(
+            role="Test Agent",
+            goal="Test goal",
+            backstory="Test backstory",
+            mcps=["http://localhost:7365/mcp"],
+        )
+        assert agent.mcps == ["http://localhost:7365/mcp"]
+
+    def test_validator_accepts_https_urls(self):
+        """Test that validator still accepts https:// URLs."""
+        agent = Agent(
+            role="Test Agent",
+            goal="Test goal",
+            backstory="Test backstory",
+            mcps=["https://api.example.com/mcp"],
+        )
+        assert agent.mcps == ["https://api.example.com/mcp"]
+
+    def test_validator_accepts_crewai_amp_urls(self):
+        """Test that validator still accepts crewai-amp: URLs."""
+        agent = Agent(
+            role="Test Agent",
+            goal="Test goal",
+            backstory="Test backstory",
+            mcps=["crewai-amp:mcp-name"],
+        )
+        assert agent.mcps == ["crewai-amp:mcp-name"]
+
+    def test_validator_accepts_http_with_fragment(self):
+        """Test that validator accepts http:// URLs with #tool fragment."""
+        agent = Agent(
+            role="Test Agent",
+            goal="Test goal",
+            backstory="Test backstory",
+            mcps=["http://localhost:7365/mcp#diff_general_info"],
+        )
+        assert agent.mcps == ["http://localhost:7365/mcp#diff_general_info"]
+
+    def test_validator_rejects_unsupported_schemes(self):
+        """Test that validator rejects unsupported URL schemes with updated error message."""
+        with pytest.raises(ValueError) as exc_info:
+            Agent(
+                role="Test Agent",
+                goal="Test goal",
+                backstory="Test backstory",
+                mcps=["ftp://example.com/mcp"],
+            )
+        
+        error_message = str(exc_info.value)
+        assert "Invalid MCP reference: ftp://example.com/mcp" in error_message
+        assert "http://" in error_message
+        assert "https://" in error_message
+        assert "crewai-amp:" in error_message
+
+
+class TestHTTPMCPRouting:
+    """Test routing of HTTP MCP URLs."""
+
+    def test_get_mcp_tools_from_string_routes_http_urls(self):
+        """Test that _get_mcp_tools_from_string routes http:// URLs correctly."""
+        agent = Agent(
+            role="Test Agent",
+            goal="Test goal",
+            backstory="Test backstory",
+        )
+        
+        sentinel_tools = [MagicMock()]
+        
+        with patch.object(agent, '_get_external_mcp_tools', return_value=sentinel_tools) as mock_external:
+            result = agent._get_mcp_tools_from_string("http://localhost:7365/mcp")
+            
+            assert result == sentinel_tools
+            mock_external.assert_called_once_with("http://localhost:7365/mcp")
+
+    def test_get_mcp_tools_from_string_routes_https_urls(self):
+        """Test that _get_mcp_tools_from_string still routes https:// URLs correctly."""
+        agent = Agent(
+            role="Test Agent",
+            goal="Test goal",
+            backstory="Test backstory",
+        )
+        
+        sentinel_tools = [MagicMock()]
+        
+        with patch.object(agent, '_get_external_mcp_tools', return_value=sentinel_tools) as mock_external:
+            result = agent._get_mcp_tools_from_string("https://api.example.com/mcp")
+            
+            assert result == sentinel_tools
+            mock_external.assert_called_once_with("https://api.example.com/mcp")
+
+
+class TestMCPServerHTTPConfig:
+    """Test MCPServerHTTP configuration with HTTP URLs."""
+
+    def test_mcp_server_http_accepts_http_url(self):
+        """Test that MCPServerHTTP accepts http:// URLs (prevent regression)."""
+        config = MCPServerHTTP(url="http://localhost:8000/mcp")
+        assert config.url == "http://localhost:8000/mcp"
+
+    def test_mcp_server_http_accepts_https_url(self):
+        """Test that MCPServerHTTP still accepts https:// URLs."""
+        config = MCPServerHTTP(url="https://api.example.com/mcp")
+        assert config.url == "https://api.example.com/mcp"
+
+    def test_agent_with_http_mcp_server_config(self):
+        """Test that Agent accepts MCPServerHTTP with http:// URL."""
+        http_config = MCPServerHTTP(url="http://localhost:8000/mcp")
+        
+        agent = Agent(
+            role="Test Agent",
+            goal="Test goal",
+            backstory="Test backstory",
+            mcps=[http_config],
+        )
+        
+        assert agent.mcps == [http_config]
+
+
+class TestHTTPMCPFragmentFiltering:
+    """Test fragment filtering for HTTP MCP URLs."""
+
+    def test_http_url_with_fragment_filters_correctly(self):
+        """Test that http:// URL with #tool fragment filters correctly."""
+        agent = Agent(
+            role="Test Agent",
+            goal="Test goal",
+            backstory="Test backstory",
+        )
+        
+        mock_schemas = {
+            "tool1": {"description": "Tool 1"},
+            "tool2": {"description": "Tool 2"},
+            "specific_tool": {"description": "Specific Tool"},
+        }
+        
+        with patch.object(agent, '_get_mcp_tool_schemas', return_value=mock_schemas):
+            with patch('crewai.tools.mcp_tool_wrapper.MCPToolWrapper') as mock_wrapper_class:
+                mock_tool = MagicMock()
+                mock_wrapper_class.return_value = mock_tool
+                
+                result = agent._get_external_mcp_tools("http://localhost:7365/mcp#specific_tool")
+                
+                mock_wrapper_class.assert_called_once()
+                call_args = mock_wrapper_class.call_args
+                assert call_args.kwargs['tool_name'] == 'specific_tool'
+                
+                assert len(result) == 1
+                assert result[0] == mock_tool
+
+
+class TestHTTPMCPWarningLog:
+    """Test warning log for HTTP MCP URLs."""
+
+    def test_warning_log_emitted_for_http_url(self):
+        """Test that warning log is emitted when http:// is used."""
+        agent = Agent(
+            role="Test Agent",
+            goal="Test goal",
+            backstory="Test backstory",
+            verbose=True,
+        )
+        
+        log_calls = []
+        logger_class = type(agent._logger)
+        original_log = logger_class.log
+        
+        def mock_log(self, level, message):
+            log_calls.append((level, message))
+            return original_log(self, level, message)
+        
+        with patch.object(logger_class, 'log', new=mock_log):
+            with patch.object(agent, '_get_mcp_tool_schemas', return_value={}):
+                agent._get_external_mcp_tools("http://localhost:7365/mcp")
+                
+                warning_messages = [msg for level, msg in log_calls if level == "warning"]
+                assert any("http://" in msg for msg in warning_messages)
+                assert any("local development" in msg for msg in warning_messages)
+                assert any("https://" in msg for msg in warning_messages)
+
+    def test_no_warning_log_for_https_url(self):
+        """Test that no warning log is emitted for https:// URLs."""
+        agent = Agent(
+            role="Test Agent",
+            goal="Test goal",
+            backstory="Test backstory",
+            verbose=True,
+        )
+        
+        log_calls = []
+        logger_class = type(agent._logger)
+        original_log = logger_class.log
+        
+        def mock_log(self, level, message):
+            log_calls.append((level, message))
+            return original_log(self, level, message)
+        
+        with patch.object(logger_class, 'log', new=mock_log):
+            with patch.object(agent, '_get_mcp_tool_schemas', return_value={}):
+                agent._get_external_mcp_tools("https://api.example.com/mcp")
+                
+                warning_messages = [msg for level, msg in log_calls if level == "warning"]
+                assert not any("http://" in msg and "local development" in msg for msg in warning_messages)