Anthropic
Email checker with Anthropic Claude SDK. Verify emails using Claude AI tool calling.
通过 Anthropic 的原生工具使用功能,直接在 Claude 模型中使用 EmailVerify。
安装
npm install @anthropic-ai/sdk基础设置
import Anthropic from '@anthropic-ai/sdk';
const anthropic = new Anthropic();
const tools: Anthropic.Tool[] = [
{
name: 'verify_email',
description: '验证邮箱地址是否有效和可送达。返回状态、可送达性、一次性邮箱标志和置信度评分。',
input_schema: {
type: 'object' as const,
properties: {
email: {
type: 'string',
description: '要验证的邮箱地址',
},
},
required: ['email'],
},
},
];实现
import Anthropic from '@anthropic-ai/sdk';
const anthropic = new Anthropic();
const tools: Anthropic.Tool[] = [
{
name: 'verify_email',
description: '验证邮箱地址是否有效和可送达。返回状态、可送达性、一次性邮箱标志和置信度评分。',
input_schema: {
type: 'object' as const,
properties: {
email: {
type: 'string',
description: '要验证的邮箱地址',
},
},
required: ['email'],
},
},
];
async function verifyEmail(email: string) {
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 response.json();
}
async function chat(userMessage: string) {
const messages: Anthropic.MessageParam[] = [
{ role: 'user', content: userMessage },
];
const response = await anthropic.messages.create({
model: 'claude-3-5-sonnet-20241022',
max_tokens: 1024,
tools,
messages,
});
// 处理工具使用
if (response.stop_reason === 'tool_use') {
const toolUseBlock = response.content.find(
(block): block is Anthropic.ToolUseBlock => block.type === 'tool_use'
);
if (toolUseBlock && toolUseBlock.name === 'verify_email') {
const result = await verifyEmail((toolUseBlock.input as { email: string }).email);
messages.push({ role: 'assistant', content: response.content });
messages.push({
role: 'user',
content: [
{
type: 'tool_result',
tool_use_id: toolUseBlock.id,
content: JSON.stringify(result),
},
],
});
const finalResponse = await anthropic.messages.create({
model: 'claude-3-5-sonnet-20241022',
max_tokens: 1024,
tools,
messages,
});
const textBlock = finalResponse.content.find(
(block): block is Anthropic.TextBlock => block.type === 'text'
);
return textBlock?.text;
}
}
const textBlock = response.content.find(
(block): block is Anthropic.TextBlock => block.type === 'text'
);
return textBlock?.text;
}
// 使用示例
const result = await chat('帮我验证 test@example.com');
console.log(result);多个工具
const tools: Anthropic.Tool[] = [
{
name: 'verify_email',
description: '验证单个邮箱地址',
input_schema: {
type: 'object' as const,
properties: {
email: {
type: 'string',
description: '要验证的邮箱地址',
},
},
required: ['email'],
},
},
{
name: 'verify_emails_bulk',
description: '一次验证多个邮箱地址',
input_schema: {
type: 'object' as const,
properties: {
emails: {
type: 'array',
items: { type: 'string' },
description: '要验证的邮箱地址数组',
},
},
required: ['emails'],
},
},
{
name: 'check_credits',
description: '检查剩余验证额度',
input_schema: {
type: 'object' as const,
properties: {},
},
},
];处理多次工具调用
Claude 可能在对话中多次调用工具:
async function chatWithMultipleTools(userMessage: string) {
let messages: Anthropic.MessageParam[] = [
{ role: 'user', content: userMessage },
];
while (true) {
const response = await anthropic.messages.create({
model: 'claude-3-5-sonnet-20241022',
max_tokens: 1024,
tools,
messages,
});
// 检查 Claude 是否想要使用工具
if (response.stop_reason === 'tool_use') {
const toolResults: Anthropic.ToolResultBlockParam[] = [];
for (const block of response.content) {
if (block.type === 'tool_use') {
let result: any;
switch (block.name) {
case 'verify_email':
result = await verifyEmail((block.input as { email: string }).email);
break;
case 'verify_emails_bulk':
result = await verifyEmailsBulk((block.input as { emails: string[] }).emails);
break;
case 'check_credits':
result = await checkCredits();
break;
}
toolResults.push({
type: 'tool_result',
tool_use_id: block.id,
content: JSON.stringify(result),
});
}
}
messages.push({ role: 'assistant', content: response.content });
messages.push({ role: 'user', content: toolResults });
} else {
// 没有更多工具调用,返回最终响应
const textBlock = response.content.find(
(block): block is Anthropic.TextBlock => block.type === 'text'
);
return textBlock?.text;
}
}
}使用流式传输
async function chatStream(userMessage: string) {
const messages: Anthropic.MessageParam[] = [
{ role: 'user', content: userMessage },
];
const stream = anthropic.messages.stream({
model: 'claude-3-5-sonnet-20241022',
max_tokens: 1024,
tools,
messages,
});
for await (const event of stream) {
if (event.type === 'content_block_delta') {
if (event.delta.type === 'text_delta') {
process.stdout.write(event.delta.text);
}
}
}
const finalMessage = await stream.finalMessage();
// 如果需要,处理工具使用
if (finalMessage.stop_reason === 'tool_use') {
// 处理工具调用...
}
}Python 示例
import anthropic
import requests
import json
import os
client = anthropic.Anthropic()
tools = [
{
"name": "verify_email",
"description": "验证邮箱地址是否有效和可送达",
"input_schema": {
"type": "object",
"properties": {
"email": {
"type": "string",
"description": "要验证的邮箱地址"
}
},
"required": ["email"]
}
}
]
def verify_email(email: str) -> dict:
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()
def chat(user_message: str) -> str:
messages = [{"role": "user", "content": user_message}]
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
tools=tools,
messages=messages
)
if response.stop_reason == "tool_use":
tool_use = next(
block for block in response.content
if block.type == "tool_use"
)
result = verify_email(tool_use.input["email"])
messages.append({"role": "assistant", "content": response.content})
messages.append({
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": tool_use.id,
"content": json.dumps(result)
}]
})
final_response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
tools=tools,
messages=messages
)
return next(
block.text for block in final_response.content
if hasattr(block, "text")
)
return next(
block.text for block in response.content
if hasattr(block, "text")
)
# 使用示例
result = chat("john@google.com 是有效的邮箱吗?")
print(result)