feat: add MCP Discovery tool for dynamic MCP server discovery

This adds a new MCPDiscoveryTool that enables CrewAI agents to dynamically
discover MCP servers based on natural language queries using the MCP
Discovery API.

Features:
- Semantic search for MCP servers using natural language
- Support for constraints (max latency, required features, excluded servers)
- Returns server recommendations with installation instructions, capabilities,
  and performance metrics
- Includes comprehensive test coverage

Closes #4249

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

lib/crewai-tools/src/crewai_tools/__init__.py

@@ -89,6 +89,9 @@ from crewai_tools.tools.jina_scrape_website_tool.jina_scrape_website_tool import
 from crewai_tools.tools.json_search_tool.json_search_tool import JSONSearchTool
 from crewai_tools.tools.linkup.linkup_search_tool import LinkupSearchTool
 from crewai_tools.tools.llamaindex_tool.llamaindex_tool import LlamaIndexTool
+from crewai_tools.tools.mcp_discovery_tool.mcp_discovery_tool import (
+    MCPDiscoveryTool,
+)
 from crewai_tools.tools.mdx_search_tool.mdx_search_tool import MDXSearchTool
 from crewai_tools.tools.merge_agent_handler_tool.merge_agent_handler_tool import (
     MergeAgentHandlerTool,
@@ -236,6 +239,7 @@ __all__ = [
     "JinaScrapeWebsiteTool",
     "LinkupSearchTool",
     "LlamaIndexTool",
+    "MCPDiscoveryTool",
     "MCPServerAdapter",
     "MDXSearchTool",
     "MergeAgentHandlerTool",

lib/crewai-tools/src/crewai_tools/tools/mcp_discovery_tool/__init__.py

@@ -0,0 +1,16 @@
+from crewai_tools.tools.mcp_discovery_tool.mcp_discovery_tool import (
+    MCPDiscoveryResult,
+    MCPDiscoveryTool,
+    MCPDiscoveryToolSchema,
+    MCPServerMetrics,
+    MCPServerRecommendation,
+)
+
+
+__all__ = [
+    "MCPDiscoveryResult",
+    "MCPDiscoveryTool",
+    "MCPDiscoveryToolSchema",
+    "MCPServerMetrics",
+    "MCPServerRecommendation",
+]

lib/crewai-tools/src/crewai_tools/tools/mcp_discovery_tool/mcp_discovery_tool.py

@@ -0,0 +1,414 @@
+"""MCP Discovery Tool for CrewAI agents.
+
+This tool enables agents to dynamically discover MCP servers based on
+natural language queries using the MCP Discovery API.
+"""
+
+import json
+import logging
+import os
+from typing import Any, TypedDict
+
+from crewai.tools import BaseTool, EnvVar
+from pydantic import BaseModel, Field
+import requests
+
+
+logger = logging.getLogger(__name__)
+
+
+class MCPServerMetrics(TypedDict, total=False):
+    """Performance metrics for an MCP server."""
+
+    avg_latency_ms: float | None
+    uptime_pct: float | None
+    last_checked: str | None
+
+
+class MCPServerRecommendation(TypedDict, total=False):
+    """A recommended MCP server from the discovery API."""
+
+    server: str
+    npm_package: str
+    install_command: str
+    confidence: float
+    description: str
+    capabilities: list[str]
+    metrics: MCPServerMetrics
+    docs_url: str
+    github_url: str
+
+
+class MCPDiscoveryResult(TypedDict):
+    """Result from the MCP Discovery API."""
+
+    recommendations: list[MCPServerRecommendation]
+    total_found: int
+    query_time_ms: int
+
+
+class MCPDiscoveryConstraints(BaseModel):
+    """Constraints for MCP server discovery."""
+
+    max_latency_ms: int | None = Field(
+        default=None,
+        description="Maximum acceptable latency in milliseconds",
+    )
+    required_features: list[str] | None = Field(
+        default=None,
+        description="List of required features/capabilities",
+    )
+    exclude_servers: list[str] | None = Field(
+        default=None,
+        description="List of server names to exclude from results",
+    )
+
+
+class MCPDiscoveryToolSchema(BaseModel):
+    """Input schema for MCPDiscoveryTool."""
+
+    need: str = Field(
+        ...,
+        description=(
+            "Natural language description of what you need. "
+            "For example: 'database with authentication', 'email automation', "
+            "'file storage', 'web scraping'"
+        ),
+    )
+    constraints: MCPDiscoveryConstraints | None = Field(
+        default=None,
+        description="Optional constraints to filter results",
+    )
+    limit: int = Field(
+        default=5,
+        description="Maximum number of recommendations to return (1-10)",
+        ge=1,
+        le=10,
+    )
+
+
+class MCPDiscoveryTool(BaseTool):
+    """Tool for discovering MCP servers dynamically.
+
+    This tool uses the MCP Discovery API to find MCP servers that match
+    a natural language description of what the agent needs. It enables
+    agents to dynamically discover and select the best MCP servers for
+    their tasks without requiring pre-configuration.
+
+    Example:
+        ```python
+        from crewai import Agent
+        from crewai_tools import MCPDiscoveryTool
+
+        discovery_tool = MCPDiscoveryTool()
+
+        agent = Agent(
+            role='Researcher',
+            tools=[discovery_tool],
+            goal='Research and analyze data'
+        )
+
+        # The agent can now discover MCP servers dynamically:
+        # discover_mcp_server(need="database with authentication")
+        # Returns: Supabase MCP server with installation instructions
+        ```
+
+    Attributes:
+        name: The name of the tool.
+        description: A description of what the tool does.
+        args_schema: The Pydantic model for input validation.
+        base_url: The base URL for the MCP Discovery API.
+        timeout: Request timeout in seconds.
+    """
+
+    name: str = "Discover MCP Server"
+    description: str = (
+        "Discover MCP (Model Context Protocol) servers that match your needs. "
+        "Use this tool to find the best MCP server for any task using natural "
+        "language. Returns server recommendations with installation instructions, "
+        "capabilities, and performance metrics."
+    )
+    args_schema: type[BaseModel] = MCPDiscoveryToolSchema
+    base_url: str = "https://mcp-discovery-production.up.railway.app"
+    timeout: int = 30
+    env_vars: list[EnvVar] = Field(
+        default_factory=lambda: [
+            EnvVar(
+                name="MCP_DISCOVERY_API_KEY",
+                description="API key for MCP Discovery (optional for free tier)",
+                required=False,
+            ),
+        ]
+    )
+
+    def _build_request_payload(
+        self,
+        need: str,
+        constraints: MCPDiscoveryConstraints | None,
+        limit: int,
+    ) -> dict[str, Any]:
+        """Build the request payload for the discovery API.
+
+        Args:
+            need: Natural language description of what is needed.
+            constraints: Optional constraints to filter results.
+            limit: Maximum number of recommendations.
+
+        Returns:
+            Dictionary containing the request payload.
+        """
+        payload: dict[str, Any] = {
+            "need": need,
+            "limit": limit,
+        }
+
+        if constraints:
+            constraints_dict: dict[str, Any] = {}
+            if constraints.max_latency_ms is not None:
+                constraints_dict["max_latency_ms"] = constraints.max_latency_ms
+            if constraints.required_features:
+                constraints_dict["required_features"] = constraints.required_features
+            if constraints.exclude_servers:
+                constraints_dict["exclude_servers"] = constraints.exclude_servers
+            if constraints_dict:
+                payload["constraints"] = constraints_dict
+
+        return payload
+
+    def _make_api_request(self, payload: dict[str, Any]) -> dict[str, Any]:
+        """Make a request to the MCP Discovery API.
+
+        Args:
+            payload: The request payload.
+
+        Returns:
+            The API response as a dictionary.
+
+        Raises:
+            ValueError: If the API returns an empty response.
+            requests.exceptions.RequestException: If the request fails.
+        """
+        url = f"{self.base_url}/api/v1/discover"
+        headers = {
+            "Content-Type": "application/json",
+        }
+
+        api_key = os.environ.get("MCP_DISCOVERY_API_KEY")
+        if api_key:
+            headers["Authorization"] = f"Bearer {api_key}"
+
+        response = None
+        try:
+            response = requests.post(
+                url,
+                headers=headers,
+                json=payload,
+                timeout=self.timeout,
+            )
+            response.raise_for_status()
+            results = response.json()
+            if not results:
+                logger.error("Empty response from MCP Discovery API")
+                raise ValueError("Empty response from MCP Discovery API")
+            return results
+        except requests.exceptions.RequestException as e:
+            error_msg = f"Error making request to MCP Discovery API: {e}"
+            if response is not None and hasattr(response, "content"):
+                error_msg += (
+                    f"\nResponse content: "
+                    f"{response.content.decode('utf-8', errors='replace')}"
+                )
+            logger.error(error_msg)
+            raise
+        except json.JSONDecodeError as e:
+            if response is not None and hasattr(response, "content"):
+                logger.error(f"Error decoding JSON response: {e}")
+                logger.error(
+                    f"Response content: "
+                    f"{response.content.decode('utf-8', errors='replace')}"
+                )
+            else:
+                logger.error(
+                    f"Error decoding JSON response: {e} (No response content available)"
+                )
+            raise
+
+    def _process_single_recommendation(
+        self, rec: dict[str, Any]
+    ) -> MCPServerRecommendation | None:
+        """Process a single recommendation from the API.
+
+        Args:
+            rec: Raw recommendation dictionary from the API.
+
+        Returns:
+            Processed MCPServerRecommendation or None if malformed.
+        """
+        try:
+            metrics_data = rec.get("metrics", {}) if isinstance(rec, dict) else {}
+            metrics: MCPServerMetrics = {
+                "avg_latency_ms": metrics_data.get("avg_latency_ms"),
+                "uptime_pct": metrics_data.get("uptime_pct"),
+                "last_checked": metrics_data.get("last_checked"),
+            }
+
+            recommendation: MCPServerRecommendation = {
+                "server": rec.get("server", ""),
+                "npm_package": rec.get("npm_package", ""),
+                "install_command": rec.get("install_command", ""),
+                "confidence": rec.get("confidence", 0.0),
+                "description": rec.get("description", ""),
+                "capabilities": rec.get("capabilities", []),
+                "metrics": metrics,
+                "docs_url": rec.get("docs_url", ""),
+                "github_url": rec.get("github_url", ""),
+            }
+            return recommendation
+        except (KeyError, TypeError, AttributeError) as e:
+            logger.warning(f"Skipping malformed recommendation: {rec}, error: {e}")
+            return None
+
+    def _process_recommendations(
+        self, recommendations: list[dict[str, Any]]
+    ) -> list[MCPServerRecommendation]:
+        """Process and validate server recommendations.
+
+        Args:
+            recommendations: Raw recommendations from the API.
+
+        Returns:
+            List of processed MCPServerRecommendation objects.
+        """
+        processed: list[MCPServerRecommendation] = []
+        for rec in recommendations:
+            result = self._process_single_recommendation(rec)
+            if result is not None:
+                processed.append(result)
+        return processed
+
+    def _format_result(self, result: MCPDiscoveryResult) -> str:
+        """Format the discovery result as a human-readable string.
+
+        Args:
+            result: The discovery result to format.
+
+        Returns:
+            A formatted string representation of the result.
+        """
+        if not result["recommendations"]:
+            return "No MCP servers found matching your requirements."
+
+        lines = [
+            f"Found {result['total_found']} MCP server(s) "
+            f"(query took {result['query_time_ms']}ms):\n"
+        ]
+
+        for i, rec in enumerate(result["recommendations"], 1):
+            confidence_pct = rec.get("confidence", 0) * 100
+            lines.append(f"{i}. {rec.get('server', 'Unknown')} ({confidence_pct:.0f}% confidence)")
+            lines.append(f"   Description: {rec.get('description', 'N/A')}")
+            lines.append(f"   Capabilities: {', '.join(rec.get('capabilities', []))}")
+            lines.append(f"   Install: {rec.get('install_command', 'N/A')}")
+            lines.append(f"   NPM Package: {rec.get('npm_package', 'N/A')}")
+
+            metrics = rec.get("metrics", {})
+            if metrics.get("avg_latency_ms") is not None:
+                lines.append(f"   Avg Latency: {metrics['avg_latency_ms']}ms")
+            if metrics.get("uptime_pct") is not None:
+                lines.append(f"   Uptime: {metrics['uptime_pct']}%")
+
+            if rec.get("docs_url"):
+                lines.append(f"   Docs: {rec['docs_url']}")
+            if rec.get("github_url"):
+                lines.append(f"   GitHub: {rec['github_url']}")
+            lines.append("")
+
+        return "\n".join(lines)
+
+    def _run(self, **kwargs: Any) -> str:
+        """Execute the MCP discovery operation.
+
+        Args:
+            **kwargs: Keyword arguments matching MCPDiscoveryToolSchema.
+
+        Returns:
+            A formatted string with discovered MCP servers.
+
+        Raises:
+            ValueError: If required parameters are missing.
+        """
+        need: str | None = kwargs.get("need")
+        if not need:
+            raise ValueError("'need' parameter is required")
+
+        constraints_data = kwargs.get("constraints")
+        constraints: MCPDiscoveryConstraints | None = None
+        if constraints_data:
+            if isinstance(constraints_data, dict):
+                constraints = MCPDiscoveryConstraints(**constraints_data)
+            elif isinstance(constraints_data, MCPDiscoveryConstraints):
+                constraints = constraints_data
+
+        limit: int = kwargs.get("limit", 5)
+
+        payload = self._build_request_payload(need, constraints, limit)
+        response = self._make_api_request(payload)
+
+        recommendations = self._process_recommendations(
+            response.get("recommendations", [])
+        )
+
+        result: MCPDiscoveryResult = {
+            "recommendations": recommendations,
+            "total_found": response.get("total_found", len(recommendations)),
+            "query_time_ms": response.get("query_time_ms", 0),
+        }
+
+        return self._format_result(result)
+
+    def discover(
+        self,
+        need: str,
+        constraints: MCPDiscoveryConstraints | None = None,
+        limit: int = 5,
+    ) -> MCPDiscoveryResult:
+        """Discover MCP servers matching the given requirements.
+
+        This is a convenience method that returns structured data instead
+        of a formatted string.
+
+        Args:
+            need: Natural language description of what is needed.
+            constraints: Optional constraints to filter results.
+            limit: Maximum number of recommendations (1-10).
+
+        Returns:
+            MCPDiscoveryResult containing server recommendations.
+
+        Example:
+            ```python
+            tool = MCPDiscoveryTool()
+            result = tool.discover(
+                need="database with authentication",
+                constraints=MCPDiscoveryConstraints(
+                    max_latency_ms=200,
+                    required_features=["auth", "realtime"]
+                ),
+                limit=3
+            )
+            for rec in result["recommendations"]:
+                print(f"{rec['server']}: {rec['description']}")
+            ```
+        """
+        payload = self._build_request_payload(need, constraints, limit)
+        response = self._make_api_request(payload)
+
+        recommendations = self._process_recommendations(
+            response.get("recommendations", [])
+        )
+
+        return {
+            "recommendations": recommendations,
+            "total_found": response.get("total_found", len(recommendations)),
+            "query_time_ms": response.get("query_time_ms", 0),
+        }

lib/crewai-tools/tests/tools/mcp_discovery_tool/__init__.py

@@ -0,0 +1 @@
+

lib/crewai-tools/tests/tools/mcp_discovery_tool/test_mcp_discovery_tool.py

@@ -0,0 +1,452 @@
+"""Tests for the MCP Discovery Tool."""
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+import requests
+
+from crewai_tools.tools.mcp_discovery_tool.mcp_discovery_tool import (
+    MCPDiscoveryConstraints,
+    MCPDiscoveryResult,
+    MCPDiscoveryTool,
+    MCPDiscoveryToolSchema,
+    MCPServerMetrics,
+    MCPServerRecommendation,
+)
+
+
+@pytest.fixture
+def mock_api_response() -> dict:
+    """Create a mock API response for testing."""
+    return {
+        "recommendations": [
+            {
+                "server": "sqlite-server",
+                "npm_package": "@modelcontextprotocol/server-sqlite",
+                "install_command": "npx -y @modelcontextprotocol/server-sqlite",
+                "confidence": 0.38,
+                "description": "SQLite database server for MCP.",
+                "capabilities": ["sqlite", "sql", "database", "embedded"],
+                "metrics": {
+                    "avg_latency_ms": 50.0,
+                    "uptime_pct": 99.9,
+                    "last_checked": "2026-01-17T10:30:00Z",
+                },
+                "docs_url": "https://modelcontextprotocol.io/docs/servers/sqlite",
+                "github_url": "https://github.com/modelcontextprotocol/servers",
+            },
+            {
+                "server": "postgres-server",
+                "npm_package": "@modelcontextprotocol/server-postgres",
+                "install_command": "npx -y @modelcontextprotocol/server-postgres",
+                "confidence": 0.33,
+                "description": "PostgreSQL database server for MCP.",
+                "capabilities": ["postgres", "sql", "database", "queries"],
+                "metrics": {
+                    "avg_latency_ms": None,
+                    "uptime_pct": None,
+                    "last_checked": None,
+                },
+                "docs_url": "https://modelcontextprotocol.io/docs/servers/postgres",
+                "github_url": "https://github.com/modelcontextprotocol/servers",
+            },
+        ],
+        "total_found": 2,
+        "query_time_ms": 245,
+    }
+
+
+@pytest.fixture
+def discovery_tool() -> MCPDiscoveryTool:
+    """Create an MCPDiscoveryTool instance for testing."""
+    return MCPDiscoveryTool()
+
+
+class TestMCPDiscoveryToolSchema:
+    """Tests for MCPDiscoveryToolSchema."""
+
+    def test_schema_with_required_fields(self) -> None:
+        """Test schema with only required fields."""
+        schema = MCPDiscoveryToolSchema(need="database server")
+        assert schema.need == "database server"
+        assert schema.constraints is None
+        assert schema.limit == 5
+
+    def test_schema_with_all_fields(self) -> None:
+        """Test schema with all fields."""
+        constraints = MCPDiscoveryConstraints(
+            max_latency_ms=200,
+            required_features=["auth", "realtime"],
+            exclude_servers=["deprecated-server"],
+        )
+        schema = MCPDiscoveryToolSchema(
+            need="database with authentication",
+            constraints=constraints,
+            limit=3,
+        )
+        assert schema.need == "database with authentication"
+        assert schema.constraints is not None
+        assert schema.constraints.max_latency_ms == 200
+        assert schema.constraints.required_features == ["auth", "realtime"]
+        assert schema.constraints.exclude_servers == ["deprecated-server"]
+        assert schema.limit == 3
+
+    def test_schema_limit_validation(self) -> None:
+        """Test that limit is validated to be between 1 and 10."""
+        with pytest.raises(ValueError):
+            MCPDiscoveryToolSchema(need="test", limit=0)
+
+        with pytest.raises(ValueError):
+            MCPDiscoveryToolSchema(need="test", limit=11)
+
+
+class TestMCPDiscoveryConstraints:
+    """Tests for MCPDiscoveryConstraints."""
+
+    def test_empty_constraints(self) -> None:
+        """Test creating empty constraints."""
+        constraints = MCPDiscoveryConstraints()
+        assert constraints.max_latency_ms is None
+        assert constraints.required_features is None
+        assert constraints.exclude_servers is None
+
+    def test_full_constraints(self) -> None:
+        """Test creating constraints with all fields."""
+        constraints = MCPDiscoveryConstraints(
+            max_latency_ms=100,
+            required_features=["feature1", "feature2"],
+            exclude_servers=["server1", "server2"],
+        )
+        assert constraints.max_latency_ms == 100
+        assert constraints.required_features == ["feature1", "feature2"]
+        assert constraints.exclude_servers == ["server1", "server2"]
+
+
+class TestMCPDiscoveryTool:
+    """Tests for MCPDiscoveryTool."""
+
+    def test_tool_initialization(self, discovery_tool: MCPDiscoveryTool) -> None:
+        """Test tool initialization with default values."""
+        assert discovery_tool.name == "Discover MCP Server"
+        assert "MCP" in discovery_tool.description
+        assert discovery_tool.base_url == "https://mcp-discovery-production.up.railway.app"
+        assert discovery_tool.timeout == 30
+
+    def test_build_request_payload_basic(
+        self, discovery_tool: MCPDiscoveryTool
+    ) -> None:
+        """Test building request payload with basic parameters."""
+        payload = discovery_tool._build_request_payload(
+            need="database server",
+            constraints=None,
+            limit=5,
+        )
+        assert payload == {"need": "database server", "limit": 5}
+
+    def test_build_request_payload_with_constraints(
+        self, discovery_tool: MCPDiscoveryTool
+    ) -> None:
+        """Test building request payload with constraints."""
+        constraints = MCPDiscoveryConstraints(
+            max_latency_ms=200,
+            required_features=["auth"],
+            exclude_servers=["old-server"],
+        )
+        payload = discovery_tool._build_request_payload(
+            need="database",
+            constraints=constraints,
+            limit=3,
+        )
+        assert payload["need"] == "database"
+        assert payload["limit"] == 3
+        assert "constraints" in payload
+        assert payload["constraints"]["max_latency_ms"] == 200
+        assert payload["constraints"]["required_features"] == ["auth"]
+        assert payload["constraints"]["exclude_servers"] == ["old-server"]
+
+    def test_build_request_payload_partial_constraints(
+        self, discovery_tool: MCPDiscoveryTool
+    ) -> None:
+        """Test building request payload with partial constraints."""
+        constraints = MCPDiscoveryConstraints(max_latency_ms=100)
+        payload = discovery_tool._build_request_payload(
+            need="test",
+            constraints=constraints,
+            limit=5,
+        )
+        assert payload["constraints"] == {"max_latency_ms": 100}
+
+    def test_process_recommendations(
+        self, discovery_tool: MCPDiscoveryTool, mock_api_response: dict
+    ) -> None:
+        """Test processing recommendations from API response."""
+        recommendations = discovery_tool._process_recommendations(
+            mock_api_response["recommendations"]
+        )
+        assert len(recommendations) == 2
+
+        first_rec = recommendations[0]
+        assert first_rec["server"] == "sqlite-server"
+        assert first_rec["npm_package"] == "@modelcontextprotocol/server-sqlite"
+        assert first_rec["confidence"] == 0.38
+        assert first_rec["capabilities"] == ["sqlite", "sql", "database", "embedded"]
+        assert first_rec["metrics"]["avg_latency_ms"] == 50.0
+        assert first_rec["metrics"]["uptime_pct"] == 99.9
+
+        second_rec = recommendations[1]
+        assert second_rec["server"] == "postgres-server"
+        assert second_rec["metrics"]["avg_latency_ms"] is None
+
+    def test_process_recommendations_with_malformed_data(
+        self, discovery_tool: MCPDiscoveryTool
+    ) -> None:
+        """Test processing recommendations with malformed data."""
+        malformed_recommendations = [
+            {"server": "valid-server", "confidence": 0.5},
+            None,
+            {"invalid": "data"},
+        ]
+        recommendations = discovery_tool._process_recommendations(
+            malformed_recommendations
+        )
+        assert len(recommendations) >= 1
+        assert recommendations[0]["server"] == "valid-server"
+
+    def test_format_result_with_recommendations(
+        self, discovery_tool: MCPDiscoveryTool
+    ) -> None:
+        """Test formatting results with recommendations."""
+        result: MCPDiscoveryResult = {
+            "recommendations": [
+                {
+                    "server": "test-server",
+                    "npm_package": "@test/server",
+                    "install_command": "npx -y @test/server",
+                    "confidence": 0.85,
+                    "description": "A test server",
+                    "capabilities": ["test", "demo"],
+                    "metrics": {
+                        "avg_latency_ms": 100.0,
+                        "uptime_pct": 99.5,
+                        "last_checked": "2026-01-17T10:00:00Z",
+                    },
+                    "docs_url": "https://example.com/docs",
+                    "github_url": "https://github.com/test/server",
+                }
+            ],
+            "total_found": 1,
+            "query_time_ms": 150,
+        }
+        formatted = discovery_tool._format_result(result)
+        assert "Found 1 MCP server(s)" in formatted
+        assert "test-server" in formatted
+        assert "85% confidence" in formatted
+        assert "A test server" in formatted
+        assert "test, demo" in formatted
+        assert "npx -y @test/server" in formatted
+        assert "100.0ms" in formatted
+        assert "99.5%" in formatted
+
+    def test_format_result_empty(self, discovery_tool: MCPDiscoveryTool) -> None:
+        """Test formatting results with no recommendations."""
+        result: MCPDiscoveryResult = {
+            "recommendations": [],
+            "total_found": 0,
+            "query_time_ms": 50,
+        }
+        formatted = discovery_tool._format_result(result)
+        assert "No MCP servers found" in formatted
+
+    @patch("crewai_tools.tools.mcp_discovery_tool.mcp_discovery_tool.requests.post")
+    def test_make_api_request_success(
+        self,
+        mock_post: MagicMock,
+        discovery_tool: MCPDiscoveryTool,
+        mock_api_response: dict,
+    ) -> None:
+        """Test successful API request."""
+        mock_response = MagicMock()
+        mock_response.json.return_value = mock_api_response
+        mock_response.raise_for_status.return_value = None
+        mock_post.return_value = mock_response
+
+        result = discovery_tool._make_api_request({"need": "database", "limit": 5})
+
+        assert result == mock_api_response
+        mock_post.assert_called_once()
+        call_args = mock_post.call_args
+        assert call_args[1]["json"] == {"need": "database", "limit": 5}
+        assert call_args[1]["timeout"] == 30
+
+    @patch("crewai_tools.tools.mcp_discovery_tool.mcp_discovery_tool.requests.post")
+    def test_make_api_request_with_api_key(
+        self,
+        mock_post: MagicMock,
+        discovery_tool: MCPDiscoveryTool,
+        mock_api_response: dict,
+    ) -> None:
+        """Test API request with API key."""
+        mock_response = MagicMock()
+        mock_response.json.return_value = mock_api_response
+        mock_response.raise_for_status.return_value = None
+        mock_post.return_value = mock_response
+
+        with patch.dict("os.environ", {"MCP_DISCOVERY_API_KEY": "test-key"}):
+            discovery_tool._make_api_request({"need": "test", "limit": 5})
+
+        call_args = mock_post.call_args
+        assert "Authorization" in call_args[1]["headers"]
+        assert call_args[1]["headers"]["Authorization"] == "Bearer test-key"
+
+    @patch("crewai_tools.tools.mcp_discovery_tool.mcp_discovery_tool.requests.post")
+    def test_make_api_request_empty_response(
+        self, mock_post: MagicMock, discovery_tool: MCPDiscoveryTool
+    ) -> None:
+        """Test API request with empty response."""
+        mock_response = MagicMock()
+        mock_response.json.return_value = {}
+        mock_response.raise_for_status.return_value = None
+        mock_post.return_value = mock_response
+
+        with pytest.raises(ValueError, match="Empty response"):
+            discovery_tool._make_api_request({"need": "test", "limit": 5})
+
+    @patch("crewai_tools.tools.mcp_discovery_tool.mcp_discovery_tool.requests.post")
+    def test_make_api_request_network_error(
+        self, mock_post: MagicMock, discovery_tool: MCPDiscoveryTool
+    ) -> None:
+        """Test API request with network error."""
+        mock_post.side_effect = requests.exceptions.ConnectionError("Network error")
+
+        with pytest.raises(requests.exceptions.ConnectionError):
+            discovery_tool._make_api_request({"need": "test", "limit": 5})
+
+    @patch("crewai_tools.tools.mcp_discovery_tool.mcp_discovery_tool.requests.post")
+    def test_make_api_request_json_decode_error(
+        self, mock_post: MagicMock, discovery_tool: MCPDiscoveryTool
+    ) -> None:
+        """Test API request with JSON decode error."""
+        mock_response = MagicMock()
+        mock_response.json.side_effect = json.JSONDecodeError("Error", "", 0)
+        mock_response.raise_for_status.return_value = None
+        mock_response.content = b"invalid json"
+        mock_post.return_value = mock_response
+
+        with pytest.raises(json.JSONDecodeError):
+            discovery_tool._make_api_request({"need": "test", "limit": 5})
+
+    @patch("crewai_tools.tools.mcp_discovery_tool.mcp_discovery_tool.requests.post")
+    def test_run_success(
+        self,
+        mock_post: MagicMock,
+        discovery_tool: MCPDiscoveryTool,
+        mock_api_response: dict,
+    ) -> None:
+        """Test successful _run execution."""
+        mock_response = MagicMock()
+        mock_response.json.return_value = mock_api_response
+        mock_response.raise_for_status.return_value = None
+        mock_post.return_value = mock_response
+
+        result = discovery_tool._run(need="database server")
+
+        assert "sqlite-server" in result
+        assert "postgres-server" in result
+        assert "Found 2 MCP server(s)" in result
+
+    @patch("crewai_tools.tools.mcp_discovery_tool.mcp_discovery_tool.requests.post")
+    def test_run_with_constraints(
+        self,
+        mock_post: MagicMock,
+        discovery_tool: MCPDiscoveryTool,
+        mock_api_response: dict,
+    ) -> None:
+        """Test _run with constraints."""
+        mock_response = MagicMock()
+        mock_response.json.return_value = mock_api_response
+        mock_response.raise_for_status.return_value = None
+        mock_post.return_value = mock_response
+
+        result = discovery_tool._run(
+            need="database",
+            constraints={"max_latency_ms": 100, "required_features": ["sql"]},
+            limit=3,
+        )
+
+        assert "sqlite-server" in result
+        call_args = mock_post.call_args
+        payload = call_args[1]["json"]
+        assert payload["constraints"]["max_latency_ms"] == 100
+        assert payload["constraints"]["required_features"] == ["sql"]
+        assert payload["limit"] == 3
+
+    def test_run_missing_need_parameter(
+        self, discovery_tool: MCPDiscoveryTool
+    ) -> None:
+        """Test _run with missing need parameter."""
+        with pytest.raises(ValueError, match="'need' parameter is required"):
+            discovery_tool._run()
+
+    @patch("crewai_tools.tools.mcp_discovery_tool.mcp_discovery_tool.requests.post")
+    def test_discover_method(
+        self,
+        mock_post: MagicMock,
+        discovery_tool: MCPDiscoveryTool,
+        mock_api_response: dict,
+    ) -> None:
+        """Test the discover convenience method."""
+        mock_response = MagicMock()
+        mock_response.json.return_value = mock_api_response
+        mock_response.raise_for_status.return_value = None
+        mock_post.return_value = mock_response
+
+        result = discovery_tool.discover(
+            need="database",
+            constraints=MCPDiscoveryConstraints(max_latency_ms=200),
+            limit=3,
+        )
+
+        assert isinstance(result, dict)
+        assert "recommendations" in result
+        assert "total_found" in result
+        assert "query_time_ms" in result
+        assert len(result["recommendations"]) == 2
+        assert result["total_found"] == 2
+
+    @patch("crewai_tools.tools.mcp_discovery_tool.mcp_discovery_tool.requests.post")
+    def test_discover_returns_structured_data(
+        self,
+        mock_post: MagicMock,
+        discovery_tool: MCPDiscoveryTool,
+        mock_api_response: dict,
+    ) -> None:
+        """Test that discover returns properly structured data."""
+        mock_response = MagicMock()
+        mock_response.json.return_value = mock_api_response
+        mock_response.raise_for_status.return_value = None
+        mock_post.return_value = mock_response
+
+        result = discovery_tool.discover(need="database")
+
+        first_rec = result["recommendations"][0]
+        assert "server" in first_rec
+        assert "npm_package" in first_rec
+        assert "install_command" in first_rec
+        assert "confidence" in first_rec
+        assert "description" in first_rec
+        assert "capabilities" in first_rec
+        assert "metrics" in first_rec
+        assert "docs_url" in first_rec
+        assert "github_url" in first_rec
+
+
+class TestMCPDiscoveryToolIntegration:
+    """Integration tests for MCPDiscoveryTool (requires network)."""
+
+    @pytest.mark.skip(reason="Integration test - requires network access")
+    def test_real_api_call(self) -> None:
+        """Test actual API call to MCP Discovery service."""
+        tool = MCPDiscoveryTool()
+        result = tool._run(need="database", limit=3)
+        assert "MCP server" in result or "No MCP servers found" in result