73b1a0feda
- Python FastAPI backend (chatbot-api/) with Claude Sonnet 4.6, prompt injection protection, rate limiting (30 msg/session), off-topic filtering, Redis session storage - Rocket.Chat integration for live monitoring and human takeover - Lead capture via n8n webhook - React chat widget: floating bubble, auto-greeting after 30s, glassmorphism chat window, mobile responsive, lazy loaded, Mixpanel analytics - Nginx proxy /api/chat → chatbot-api:8000 - Docker: chatbot-api + Redis services added to docker-compose Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
67 lines
2.5 KiB
Python
67 lines
2.5 KiB
Python
import anthropic
|
|
import httpx
|
|
from config import settings
|
|
from knowledge import SYSTEM_PROMPT, TOOLS
|
|
|
|
client = anthropic.Anthropic(api_key=settings.anthropic_api_key)
|
|
|
|
|
|
async def get_ai_response(messages: list[dict]) -> tuple[str, dict | None]:
|
|
"""Get response from Claude. Returns (text_reply, lead_data_or_none)."""
|
|
response = client.messages.create(
|
|
model=settings.model,
|
|
max_tokens=settings.max_response_tokens,
|
|
system=SYSTEM_PROMPT,
|
|
tools=TOOLS,
|
|
messages=messages,
|
|
)
|
|
|
|
text_parts = []
|
|
lead_data = None
|
|
|
|
for block in response.content:
|
|
if block.type == "text":
|
|
text_parts.append(block.text)
|
|
elif block.type == "tool_use" and block.name == "capture_lead":
|
|
lead_data = block.input
|
|
# Send lead to n8n webhook
|
|
try:
|
|
async with httpx.AsyncClient() as http:
|
|
await http.post(
|
|
f"{settings.n8n_webhook_url}/chatbot-lead",
|
|
json=lead_data,
|
|
timeout=10,
|
|
)
|
|
except Exception:
|
|
pass # Don't fail the chat if webhook fails
|
|
|
|
reply = " ".join(text_parts) if text_parts else "Thank you! I've noted your details. We'll be in touch shortly."
|
|
|
|
# If there was a tool use but no text, we need to send tool result back to get text
|
|
if not text_parts and any(b.type == "tool_use" for b in response.content):
|
|
tool_block = next(b for b in response.content if b.type == "tool_use")
|
|
followup = client.messages.create(
|
|
model=settings.model,
|
|
max_tokens=settings.max_response_tokens,
|
|
system=SYSTEM_PROMPT,
|
|
tools=TOOLS,
|
|
messages=messages + [
|
|
{"role": "assistant", "content": response.content},
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "tool_result",
|
|
"tool_use_id": tool_block.id,
|
|
"content": "Lead captured successfully. Thank the visitor and offer next steps.",
|
|
}
|
|
],
|
|
},
|
|
],
|
|
)
|
|
for block in followup.content:
|
|
if block.type == "text":
|
|
text_parts.append(block.text)
|
|
reply = " ".join(text_parts) if text_parts else "Thank you! I've noted your details. We'll be in touch shortly."
|
|
|
|
return reply, lead_data
|