enrich_organization
Enhance company profiles by retrieving detailed organization data including industry classification, employee count, and contact information using domain-based lookup.
Instructions
Enrich organization/company information.
This tool enriches company data based on domain, returning detailed information about the organization including industry, employee count, contact info, and more.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| request | Yes |
Implementation Reference
- src/apollo_mcp_server.py:223-255 (handler)The enrich_organization tool handler: validates input using schema, makes GET request to Apollo.io /v1/organizations/enrich endpoint with domain param, returns enriched organization data or error.@mcp.tool() async def enrich_organization(request: Union[Dict[str, Any], str]) -> Dict[str, Any]: """ Enrich organization/company information. This tool enriches company data based on domain, returning detailed information about the organization including industry, employee count, contact info, and more. """ endpoint = "/v1/organizations/enrich" # 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: org_request = OrganizationEnrichmentRequest(**request) except Exception as e: return {"error": f"Invalid request parameters: {str(e)}"} params = {"domain": org_request.domain} try: result = await apollo_client.make_request("GET", endpoint, params=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:112-116 (schema)Pydantic schema for enrich_organization input: requires 'domain' string field.class OrganizationEnrichmentRequest(BaseModel): """Request model for organization enrichment.""" domain: str = Field(..., description="Company domain to enrich")
- src/apollo_mcp_server.py:223-223 (registration)FastMCP @mcp.tool() decorator registers the enrich_organization function as an MCP tool.@mcp.tool()
- src/apollo_mcp_server.py:50-72 (helper)ApolloAPIClient.make_request helper method used by enrich_organization to call the API.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()