Apollo.io MCP Server

Instructions

Implementation Reference

• src/apollo_mcp_server.py:118-151 (handler)

  The handler function for the 'search_accounts' tool. It validates input using AccountSearchRequest, constructs API parameters, and makes a POST request to Apollo.io's /v1/accounts/search endpoint using the ApolloAPIClient.
  @mcp.tool() async def search_accounts(request: Union[Dict[str, Any], str]) -> Dict[str, Any]: """ Search for accounts/companies in Apollo.io database. This tool allows you to search for companies by name, location, employee count, and other criteria. Useful for finding potential prospects and account information. """ endpoint = "/v1/accounts/search" # Handle both JSON string and dict inputs if isinstance(request, str): try: request = json.loads(request) except json.JSONDecodeError as e: return {"error": f"Invalid JSON in request: {str(e)}"} # Create and validate request object from dictionary try: account_request = AccountSearchRequest(**request) except Exception as e: return {"error": f"Invalid request parameters: {str(e)}"} # Convert request to dict and remove None values search_params = {k: v for k, v in account_request.dict().items() if v is not None} try: result = await apollo_client.make_request("POST", endpoint, data=search_params) return result except httpx.HTTPStatusError as e: return {"error": f"API request failed: {e.response.status_code} {e.response.text}"} except Exception as e: return {"error": f"Request failed: {str(e)}"}
• src/apollo_mcp_server.py:79-87 (schema)

  Pydantic BaseModel defining the input schema/validation for the search_accounts tool parameters.
  class AccountSearchRequest(BaseModel): """Request model for account search.""" q_organization_name: Optional[str] = Field(None, description="Company name to search for") organization_locations: Optional[List[str]] = Field(None, description="List of locations to filter by") organization_num_employees_ranges: Optional[List[str]] = Field(None, description="Employee count ranges") industry_tag_ids: Optional[List[str]] = Field(None, description="Industry tag IDs") page: int = Field(1, description="Page number for pagination") per_page: int = Field(25, description="Number of results per page")
• src/apollo_mcp_server.py:38-72 (helper)

  Helper class ApolloAPIClient that handles authenticated HTTP requests to the Apollo.io API, used by the search_accounts handler.
  class ApolloAPIClient: """Apollo.io API client for making authenticated requests.""" def __init__(self, api_key: str): self.api_key = api_key self.base_url = APOLLO_BASE_URL self.headers = { "X-Api-Key": api_key, "Content-Type": "application/json", "Cache-Control": "no-cache" } async def make_request( self, method: str, endpoint: str, params: Optional[Dict] = None, data: Optional[Dict] = None ) -> Dict[str, Any]: """Make an authenticated request to the Apollo.io API.""" url = f"{self.base_url}{endpoint}" async with httpx.AsyncClient() as client: if method.upper() == "GET": response = await client.get(url, headers=self.headers, params=params) elif method.upper() == "POST": response = await client.post(url, headers=self.headers, json=data) elif method.upper() == "PUT": response = await client.put(url, headers=self.headers, json=data) else: raise ValueError(f"Unsupported HTTP method: {method}") response.raise_for_status() return response.json()
• src/apollo_mcp_server.py:118-118 (registration)

  The @mcp.tool() decorator registers the search_accounts function with the FastMCP server.
  @mcp.tool()