Cold-Leads-Tools für CrewAI- und LangChain-Agenten

# coldleads_tools.py - shared by the CrewAI and LangChain examples (pip install httpx)
import json
import os

import httpx

VERIFY_DESC = (
    "Verify one e-mail address with Cold Leads (costs 1 credit). Returns status valid, risky or invalid, "
    "a score from 0 to 100 and reason codes. The last reason says what was checked: ok means a mail server "
    "accepted the mailbox, smtp_unreachable means the mailbox itself was not checked, catch_all means the "
    "domain accepts any address, mailbox_missing, no_mx and syntax mean the address is invalid."
)
FIND_DESC = (
    "Guess a person's work e-mail address from first name, last name and company domain (costs 1 credit). "
    "Returns the most likely address, method (verified only when a mail server confirmed the mailbox, "
    "otherwise pattern), confidence from 0 to 100 and candidates. Treat a pattern result as unconfirmed."
)
SEARCH_DESC = (
    "Search the contacts already saved in the user's own Cold Leads CRM for a company domain (free). "
    "role is an optional keyword matched against name, e-mail local part, tags, notes and contact type. "
    "This is not a people database: it only returns contacts the user already has. "
    "Never e-mail contacts whose do_not_contact is true."
)


class ColdLeadsClient:
    def __init__(self, api_key=None, base_url="https://coldleads.app/api/v1"):
        self.http = httpx.Client(
            base_url=base_url,
            headers={"x-api-key": api_key or os.environ["COLDLEADS_API_KEY"]},
            timeout=40.0,
        )

    def _call(self, method, path, **kwargs) -> str:
        r = self.http.request(method, path, **kwargs)
        try:
            data = r.json()
        except ValueError:
            data = {}
        if r.status_code >= 400:
            # hand API errors (no_credits, rate_limited with retry_after_seconds, ...) to the model as data
            return json.dumps({"http_status": r.status_code, "error": f"http_{r.status_code}", **data})
        return json.dumps(data)

    def verify(self, email: str) -> str:
        return self._call("POST", "/verify", json={"email": email, "timeout_ms": 10000})

    def find(self, first: str, last: str, domain: str) -> str:
        return self._call("POST", "/find", json={"first": first, "last": last, "domain": domain})

    def search_crm(self, domain: str, role: str = "", limit: int = 10) -> str:
        params = {"domain": domain, "limit": limit}
        if role:
            params["role"] = role
        return self._call("GET", "/leads", params=params)

# pip install crewai httpx   (CrewAI needs Python 3.10-3.13)
from crewai import LLM, Agent, Crew, Task
from crewai.tools import BaseTool, EnvVar
from pydantic import BaseModel, Field

from coldleads_tools import FIND_DESC, SEARCH_DESC, VERIFY_DESC, ColdLeadsClient

client = ColdLeadsClient()
KEY = [EnvVar(name="COLDLEADS_API_KEY", description="Cold Leads secret API key (sk_...)", required=True)]


class VerifyInput(BaseModel):
    email: str = Field(..., description="E-mail address to verify, e.g. anna@example.com")


class FindInput(BaseModel):
    first: str = Field(..., description="First name")
    last: str = Field(..., description="Last name")
    domain: str = Field(..., description="Company domain, e.g. example.com")


class SearchInput(BaseModel):
    domain: str = Field(..., description="Company domain, e.g. example.com")
    role: str = Field("", description="Optional keyword such as ceo, sales or marketing")
    limit: int = Field(10, ge=1, le=50, description="Maximum number of contacts")


class VerifyEmailTool(BaseTool):
    name: str = "verify_email"
    description: str = VERIFY_DESC
    args_schema: type[BaseModel] = VerifyInput
    env_vars: list[EnvVar] = KEY

    def _run(self, email: str) -> str:
        return client.verify(email)


class FindEmailTool(BaseTool):
    name: str = "find_email"
    description: str = FIND_DESC
    args_schema: type[BaseModel] = FindInput
    env_vars: list[EnvVar] = KEY

    def _run(self, first: str, last: str, domain: str) -> str:
        return client.find(first, last, domain)


class SearchCrmTool(BaseTool):
    name: str = "search_crm_contacts"
    description: str = SEARCH_DESC
    args_schema: type[BaseModel] = SearchInput
    env_vars: list[EnvVar] = KEY

    def _run(self, domain: str, role: str = "", limit: int = 10) -> str:
        return client.search_crm(domain, role, limit)


researcher = Agent(
    role="B2B outreach researcher",
    goal="Prepare a short list of contacts whose addresses were checked with Cold Leads",
    backstory="You never invent contacts or addresses and you report what each check actually confirmed.",
    tools=[SearchCrmTool(), VerifyEmailTool(), FindEmailTool()],
    llm=LLM(model="anthropic/claude-opus-5-5", max_tokens=4096),  # any tool-calling model CrewAI supports
)
task = Task(
    description=(
        "Find the contacts at {domain} in my Cold Leads CRM whose role keyword matches {role}. "
        "Skip anyone with do_not_contact true, verify the others and report status and the last reason code."
    ),
    expected_output="A table with e-mail, name, status, score and last reason code.",
    agent=researcher,
)

if __name__ == "__main__":
    result = Crew(agents=[researcher], tasks=[task]).kickoff(inputs={"domain": "example.com", "role": "cto"})
    print(result.raw)

# pip install -U "langchain[anthropic]" httpx   (LangChain 1.x, Python 3.10+)
from langchain.agents import create_agent
from langchain.tools import tool
from pydantic import BaseModel, Field

from coldleads_tools import FIND_DESC, SEARCH_DESC, VERIFY_DESC, ColdLeadsClient

client = ColdLeadsClient()


class VerifyInput(BaseModel):
    email: str = Field(description="E-mail address to verify, e.g. anna@example.com")


class FindInput(BaseModel):
    first: str = Field(description="First name")
    last: str = Field(description="Last name")
    domain: str = Field(description="Company domain, e.g. example.com")


class SearchInput(BaseModel):
    domain: str = Field(description="Company domain, e.g. example.com")
    role: str = Field(default="", description="Optional keyword such as ceo, sales or marketing")
    limit: int = Field(default=10, ge=1, le=50, description="Maximum number of contacts")


@tool("verify_email", description=VERIFY_DESC, args_schema=VerifyInput)
def verify_email(email: str) -> str:
    return client.verify(email)


@tool("find_email", description=FIND_DESC, args_schema=FindInput)
def find_email(first: str, last: str, domain: str) -> str:
    return client.find(first, last, domain)


@tool("search_crm_contacts", description=SEARCH_DESC, args_schema=SearchInput)
def search_crm_contacts(domain: str, role: str = "", limit: int = 10) -> str:
    return client.search_crm(domain, role, limit)


agent = create_agent(
    model="anthropic:claude-opus-5-5",  # any tool-calling chat model in provider:model form
    tools=[search_crm_contacts, verify_email, find_email],
    system_prompt=(
        "You help with B2B outreach research. Use only the Cold Leads tools for contact data and never invent "
        "addresses. For every verification, report the status and the last reason code."
    ),
)

if __name__ == "__main__":
    result = agent.invoke({"messages": [{"role": "user", "content": "List my CRM contacts at example.com tagged cto and verify their addresses."}]})
    final = result["messages"][-1]
    print("".join(block["text"] for block in final.content_blocks if block["type"] == "text"))