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://encrata.com/api/agent"
HEADERS = {
    "Authorization": f"Bearer {ENCRATA_API_KEY}",
    "Content-Type": "application/json",
}

@component
class EncrtaEmailLookup:
    """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}/lookup",
                headers=HEADERS,
                json={"email": email},
            )
            resp.raise_for_status()
            results.append(resp.json())
        return {"results": results}


@component
class EncrtaEmailValidator:
    """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}/validate",
                headers=HEADERS,
                json={"email": email},
            )
            resp.raise_for_status()
            results.append(resp.json())
        return {"results": results}


@component
class EncrtaBreachChecker:
    """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", EncrtaEmailLookup())

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:
            lines.append(f"Name: {r.get('n', 'N/A')}")
            lines.append(f"Company: {r.get('co', 'N/A')}")
            lines.append(f"Role: {r.get('role', 'N/A')}")
            lines.append(f"Location: {r.get('loc', 'N/A')}")
            lines.append(f"Industry: {r.get('ind', 'N/A')}")
            lines.append(f"Validity: {r.get('v', '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", EncrtaEmailLookup())
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", EncrtaEmailLookup())

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

for i, person in enumerate(result["enricher"]["results"]):
    print(f"{emails[i]}: {person.get('n', 'N/A')} {person.get('co', 'N/A')}")