LangChain
Email checker with LangChain. Verify emails in LangChain agents and chains.
Integrate email verification as a custom tool in LangChain for both Python and JavaScript/TypeScript.
Python
Installation
pip install langchain langchain-openai requestsBasic Tool Definition
from langchain.tools import tool
import requests
import os
@tool
def verify_email(email: str) -> dict:
"""Verify if an email address is valid and deliverable.
Args:
email: The email address to verify
Returns:
Verification result including status, deliverability, and risk flags
"""
response = requests.post(
'https://api.emailverify.ai/v1/verify',
headers={
'Authorization': f'Bearer {os.environ["EMAILVERIFY_API_KEY"]}',
'Content-Type': 'application/json',
},
json={'email': email}
)
return response.json()Using with an Agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
import requests
import os
@tool
def verify_email(email: str) -> dict:
"""Verify if an email address is valid and deliverable.
Args:
email: The email address to verify
Returns:
Verification result including status, deliverability, and risk flags
"""
response = requests.post(
'https://api.emailverify.ai/v1/verify',
headers={
'Authorization': f'Bearer {os.environ["EMAILVERIFY_API_KEY"]}',
'Content-Type': 'application/json',
},
json={'email': email}
)
return response.json()
# Create agent
llm = ChatOpenAI(model="gpt-4-turbo")
tools = [verify_email]
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant that can verify email addresses."),
MessagesPlaceholder("chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad"),
])
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)
# Use the agent
result = agent_executor.invoke({
"input": "Is john@google.com a valid email?"
})
print(result["output"])Bulk Verification Tool
@tool
def verify_emails_bulk(emails: list[str]) -> dict:
"""Verify multiple email addresses at once.
Args:
emails: List of email addresses to verify (max 100)
Returns:
Bulk verification results
"""
response = requests.post(
'https://api.emailverify.ai/v1/verify/bulk',
headers={
'Authorization': f'Bearer {os.environ["EMAILVERIFY_API_KEY"]}',
'Content-Type': 'application/json',
},
json={'emails': emails[:100]}
)
return response.json()With Error Handling
@tool
def verify_email(email: str) -> dict:
"""Verify if an email address is valid and deliverable.
Args:
email: The email address to verify
Returns:
Verification result or error message
"""
try:
response = requests.post(
'https://api.emailverify.ai/v1/verify',
headers={
'Authorization': f'Bearer {os.environ["EMAILVERIFY_API_KEY"]}',
'Content-Type': 'application/json',
},
json={'email': email},
timeout=30
)
if response.status_code == 429:
return {'error': 'Rate limit exceeded. Please try again later.'}
response.raise_for_status()
return response.json()
except requests.exceptions.Timeout:
return {'error': 'Request timed out. Please try again.'}
except requests.exceptions.RequestException as e:
return {'error': f'Request failed: {str(e)}'}JavaScript/TypeScript
Installation
npm install @langchain/core @langchain/openai langchain zodBasic Tool Definition
import { DynamicStructuredTool } from '@langchain/core/tools';
import { z } from 'zod';
const verifyEmailTool = new DynamicStructuredTool({
name: 'verify_email',
description: 'Verify if an email address is valid and deliverable',
schema: z.object({
email: z.string().email().describe('The email address to verify'),
}),
func: async ({ email }) => {
const response = await fetch('https://api.emailverify.ai/v1/verify', {
method: 'POST',
headers: {
'Authorization': `Bearer ${process.env.EMAILVERIFY_API_KEY}`,
'Content-Type': 'application/json',
},
body: JSON.stringify({ email }),
});
return JSON.stringify(await response.json());
},
});Using with an Agent
import { DynamicStructuredTool } from '@langchain/core/tools';
import { ChatOpenAI } from '@langchain/openai';
import { AgentExecutor, createOpenAIFunctionsAgent } from 'langchain/agents';
import { ChatPromptTemplate, MessagesPlaceholder } from '@langchain/core/prompts';
import { z } from 'zod';
const verifyEmailTool = new DynamicStructuredTool({
name: 'verify_email',
description: 'Verify if an email address is valid and deliverable',
schema: z.object({
email: z.string().email().describe('The email address to verify'),
}),
func: async ({ email }) => {
const response = await fetch('https://api.emailverify.ai/v1/verify', {
method: 'POST',
headers: {
'Authorization': `Bearer ${process.env.EMAILVERIFY_API_KEY}`,
'Content-Type': 'application/json',
},
body: JSON.stringify({ email }),
});
return JSON.stringify(await response.json());
},
});
const llm = new ChatOpenAI({ modelName: 'gpt-4-turbo' });
const tools = [verifyEmailTool];
const prompt = ChatPromptTemplate.fromMessages([
['system', 'You are a helpful assistant that can verify email addresses.'],
new MessagesPlaceholder('chat_history'),
['human', '{input}'],
new MessagesPlaceholder('agent_scratchpad'),
]);
const agent = await createOpenAIFunctionsAgent({ llm, tools, prompt });
const agentExecutor = new AgentExecutor({ agent, tools });
const result = await agentExecutor.invoke({
input: 'Check if test@example.com is valid',
chat_history: [],
});
console.log(result.output);With Streaming
import { AgentExecutor, createOpenAIFunctionsAgent } from 'langchain/agents';
// ... setup code ...
const stream = await agentExecutor.stream({
input: 'Verify john@google.com for me',
chat_history: [],
});
for await (const chunk of stream) {
if (chunk.output) {
console.log(chunk.output);
}
}