Apollo.io MCP Server

Instructions

Implementation Reference

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

  The core handler function for the 'search_accounts' tool. It handles input parsing and validation, calls the Apollo.io API endpoint /v1/accounts/search, and returns the results or error messages.
  @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 model that defines and validates the input schema for the search_accounts tool.
  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:118-118 (registration)

  The @mcp.tool() decorator registers the search_accounts function as an MCP tool.
  @mcp.tool()