Haystack

pip install haystack-ai requests

import requests
from typing import List, Optional
from haystack import component

ENCRATA_API_KEY = "YOUR_API_KEY"
BASE_URL = "https://developer.encrata.com/api/lookup/email"
HEADERS = {
    "Authorization": f"Bearer {ENCRATA_API_KEY}",
    "Content-Type": "application/json",
}

@component
class EncrataEmailLookup:
    """Looks up detailed intelligence about people by their email addresses."""

    @component.output_types(results=List[dict])
    def run(self, emails: List[str]):
        results = []
        for email in emails:
            resp = requests.post(
                f"{BASE_URL}/identity",
                headers=HEADERS,
                json={"email": email},
            )
            resp.raise_for_status()
            results.append(resp.json())
        return {"results": results}


@component
class EncrataEmailValidator:
    """Validates whether email addresses are deliverable."""

    @component.output_types(results=List[dict])
    def run(self, emails: List[str]):
        results = []
        for email in emails:
            resp = requests.post(
                f"{BASE_URL}/validity",
                headers=HEADERS,
                json={"email": email},
            )
            resp.raise_for_status()
            results.append(resp.json())
        return {"results": results}


@component
class EncrataBreachChecker:
    """Checks emails against known data breaches."""

    @component.output_types(results=List[dict])
    def run(self, emails: List[str]):
        results = []
        for email in emails:
            resp = requests.post(
                f"{BASE_URL}/breaches",
                headers=HEADERS,
                json={"email": email},
            )
            resp.raise_for_status()
            results.append(resp.json())
        return {"results": results}

from haystack import Pipeline

pipeline = Pipeline()
pipeline.add_component("enricher", EncrataEmailLookup())

result = pipeline.run({"enricher": {"emails": ["satya@microsoft.com"]}})
print(result["enricher"]["results"][0])

from haystack import Pipeline
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import PromptBuilder

@component
class FormatEnrichment:
    """Formats enrichment results into a prompt-friendly string."""

    @component.output_types(text=str)
    def run(self, results: List[dict]):
        lines = []
        for r in results:
            p = r.get("person", {})
            lines.append(f"Name: {p.get('first_name', 'N/A')} {p.get('last_name', '')}".strip())
            lines.append(f"Company: {p.get('company', 'N/A')}")
            lines.append(f"Role: {p.get('job_role', 'N/A')}")
            lines.append(f"Location: {p.get('current_location', 'N/A')}")
            lines.append(f"Industry: {p.get('industry', 'N/A')}")
            lines.append("")
        return {"text": "\n".join(lines)}

template = """Based on this enrichment data, write a brief professional summary:

{{ text }}

Summary:"""

pipeline = Pipeline()
pipeline.add_component("enricher", EncrataEmailLookup())
pipeline.add_component("formatter", FormatEnrichment())
pipeline.add_component("prompt", PromptBuilder(template=template))
pipeline.add_component("llm", OpenAIGenerator(model="gpt-4o-mini"))

pipeline.connect("enricher.results", "formatter.results")
pipeline.connect("formatter.text", "prompt.text")
pipeline.connect("prompt.prompt", "llm.prompt")

result = pipeline.run({"enricher": {"emails": ["satya@microsoft.com"]}})
print(result["llm"]["replies"][0])

emails = [
    "satya@microsoft.com",
    "tim@apple.com",
    "sundar@google.com",
]

pipeline = Pipeline()
pipeline.add_component("enricher", EncrataEmailLookup())

result = pipeline.run({"enricher": {"emails": emails}})

for i, result_item in enumerate(result["enricher"]["results"]):
    p = result_item.get("person", {})
    print(f"{emails[i]}: {p.get('first_name', 'N/A')} {p.get('company', 'N/A')}")