Implement Python email agent with multi-model AI support
Adds a complete email-based registration agent for Spielgruppe Pumuckl based on the OpenSpec define-project-scope specifications. Architecture - Channel-agnostic EmailAgent core — no email-specific code in business logic - Pluggable AI provider layer: Anthropic (Claude) and OpenAI (GPT) supported via a shared LLMProvider interface; switch with AI_PROVIDER env var - IMAP polling for inbound emails with thread-tracking via email headers (Message-ID / In-Reply-To / References) - SMTP for outbound replies and admin notifications - File-based JSON storage for conversation state and completed registrations - Admin-editable knowledge-base loaded from markdown files at startup Key files src/config.py — env-var configuration src/providers/base.py — abstract LLMProvider src/providers/anthropic_provider.py — Claude backend src/providers/openai_provider.py — OpenAI backend src/agent/core.py — EmailAgent orchestrator src/agent/prompts.py — system prompt builder (KB + registration state) src/models/registration.py — RegistrationData matching the JSON schema src/models/conversation.py — ConversationState persisted per thread src/channels/email_channel.py — IMAP/SMTP I/O + quoted-text stripping src/storage/json_store.py — conversation & registration persistence src/notifications/notifier.py — admin notification routing by playgroup type main.py — polling entry point requirements.txt — anthropic, openai, python-dotenv, jsonschema .env.example — configuration template https://claude.ai/code/session_01HaUFs7SaLD5SoiuGCY27Tw
This commit is contained in:
Claude
@@ -0,0 +1,76 @@
|
||||
"""Conversation state model — persisted per email thread."""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from typing import Optional
|
||||
|
||||
from .registration import RegistrationData
|
||||
|
||||
|
||||
def _now() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChatMessage:
|
||||
role: str # "user" or "assistant"
|
||||
content: str
|
||||
timestamp: str = field(default_factory=_now)
|
||||
|
||||
|
||||
@dataclass
|
||||
class ConversationState:
|
||||
conversation_id: str
|
||||
language: str = "de" # "de" or "en"
|
||||
flow_step: str = "greeting" # current step in registration flow
|
||||
registration: RegistrationData = field(default_factory=RegistrationData)
|
||||
messages: list = field(default_factory=list) # list[ChatMessage]
|
||||
parent_email: str = ""
|
||||
parent_name: Optional[str] = None
|
||||
created_at: str = field(default_factory=_now)
|
||||
updated_at: str = field(default_factory=_now)
|
||||
last_activity: str = field(default_factory=_now)
|
||||
completed: bool = False
|
||||
reminder_count: int = 0
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"conversation_id": self.conversation_id,
|
||||
"language": self.language,
|
||||
"flow_step": self.flow_step,
|
||||
"registration": self.registration.to_dict(),
|
||||
"messages": [
|
||||
{"role": m.role, "content": m.content, "timestamp": m.timestamp}
|
||||
for m in self.messages
|
||||
],
|
||||
"parent_email": self.parent_email,
|
||||
"parent_name": self.parent_name,
|
||||
"created_at": self.created_at,
|
||||
"updated_at": self.updated_at,
|
||||
"last_activity": self.last_activity,
|
||||
"completed": self.completed,
|
||||
"reminder_count": self.reminder_count,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "ConversationState":
|
||||
state = cls(conversation_id=data["conversation_id"])
|
||||
state.language = data.get("language", "de")
|
||||
state.flow_step = data.get("flow_step", "greeting")
|
||||
state.registration = RegistrationData.from_dict(data.get("registration", {}))
|
||||
state.messages = [
|
||||
ChatMessage(
|
||||
role=m["role"],
|
||||
content=m["content"],
|
||||
timestamp=m.get("timestamp", ""),
|
||||
)
|
||||
for m in data.get("messages", [])
|
||||
]
|
||||
state.parent_email = data.get("parent_email", "")
|
||||
state.parent_name = data.get("parent_name")
|
||||
state.created_at = data.get("created_at", "")
|
||||
state.updated_at = data.get("updated_at", "")
|
||||
state.last_activity = data.get("last_activity", "")
|
||||
state.completed = data.get("completed", False)
|
||||
state.reminder_count = data.get("reminder_count", 0)
|
||||
return state
|
||||
@@ -0,0 +1,126 @@
|
||||
"""Registration data models matching the JSON schema in registration-schema.json."""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class BookingDay:
|
||||
day: str # "monday", "wednesday", "thursday"
|
||||
type: str # "indoor", "outdoor"
|
||||
|
||||
|
||||
@dataclass
|
||||
class Booking:
|
||||
playgroup_types: list = field(default_factory=list) # ["indoor", "outdoor"]
|
||||
selected_days: list = field(default_factory=list) # list[BookingDay]
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChildInfo:
|
||||
full_name: Optional[str] = None
|
||||
date_of_birth: Optional[str] = None # YYYY-MM-DD
|
||||
special_needs: Optional[str] = None # text or "None"
|
||||
|
||||
|
||||
@dataclass
|
||||
class ParentGuardian:
|
||||
full_name: Optional[str] = None
|
||||
street_address: Optional[str] = None
|
||||
postal_code: Optional[str] = None # 4-digit Swiss code
|
||||
city: Optional[str] = None
|
||||
phone: Optional[str] = None
|
||||
email: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class EmergencyContact:
|
||||
full_name: Optional[str] = None
|
||||
phone: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class RegistrationData:
|
||||
child: ChildInfo = field(default_factory=ChildInfo)
|
||||
parent_guardian: ParentGuardian = field(default_factory=ParentGuardian)
|
||||
emergency_contact: EmergencyContact = field(default_factory=EmergencyContact)
|
||||
booking: Booking = field(default_factory=Booking)
|
||||
|
||||
def is_complete(self) -> bool:
|
||||
"""Return True when all required schema fields are present."""
|
||||
return (
|
||||
bool(self.child.full_name)
|
||||
and bool(self.child.date_of_birth)
|
||||
and self.child.special_needs is not None
|
||||
and bool(self.parent_guardian.full_name)
|
||||
and bool(self.parent_guardian.street_address)
|
||||
and bool(self.parent_guardian.postal_code)
|
||||
and bool(self.parent_guardian.city)
|
||||
and bool(self.parent_guardian.phone)
|
||||
and bool(self.parent_guardian.email)
|
||||
and bool(self.emergency_contact.full_name)
|
||||
and bool(self.emergency_contact.phone)
|
||||
and len(self.booking.playgroup_types) > 0
|
||||
and len(self.booking.selected_days) > 0
|
||||
)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"child": {
|
||||
"fullName": self.child.full_name,
|
||||
"dateOfBirth": self.child.date_of_birth,
|
||||
"specialNeeds": self.child.special_needs,
|
||||
},
|
||||
"parentGuardian": {
|
||||
"fullName": self.parent_guardian.full_name,
|
||||
"streetAddress": self.parent_guardian.street_address,
|
||||
"postalCode": self.parent_guardian.postal_code,
|
||||
"city": self.parent_guardian.city,
|
||||
"phone": self.parent_guardian.phone,
|
||||
"email": self.parent_guardian.email,
|
||||
},
|
||||
"emergencyContact": {
|
||||
"fullName": self.emergency_contact.full_name,
|
||||
"phone": self.emergency_contact.phone,
|
||||
},
|
||||
"booking": {
|
||||
"playgroupTypes": self.booking.playgroup_types,
|
||||
"selectedDays": [
|
||||
{"day": d.day, "type": d.type}
|
||||
for d in self.booking.selected_days
|
||||
],
|
||||
},
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "RegistrationData":
|
||||
reg = cls()
|
||||
if child := data.get("child", {}):
|
||||
reg.child = ChildInfo(
|
||||
full_name=child.get("fullName"),
|
||||
date_of_birth=child.get("dateOfBirth"),
|
||||
special_needs=child.get("specialNeeds"),
|
||||
)
|
||||
if parent := data.get("parentGuardian", {}):
|
||||
reg.parent_guardian = ParentGuardian(
|
||||
full_name=parent.get("fullName"),
|
||||
street_address=parent.get("streetAddress"),
|
||||
postal_code=parent.get("postalCode"),
|
||||
city=parent.get("city"),
|
||||
phone=parent.get("phone"),
|
||||
email=parent.get("email"),
|
||||
)
|
||||
if emergency := data.get("emergencyContact", {}):
|
||||
reg.emergency_contact = EmergencyContact(
|
||||
full_name=emergency.get("fullName"),
|
||||
phone=emergency.get("phone"),
|
||||
)
|
||||
if booking := data.get("booking", {}):
|
||||
reg.booking = Booking(
|
||||
playgroup_types=booking.get("playgroupTypes", []),
|
||||
selected_days=[
|
||||
BookingDay(day=d["day"], type=d["type"])
|
||||
for d in booking.get("selectedDays", [])
|
||||
],
|
||||
)
|
||||
return reg
|
||||
Reference in New Issue
Block a user