Structured Email Extraction with Instructor

import instructor
import requests
from openai import OpenAI
from pydantic import BaseModel, EmailStr, field_validator

MULTIMAIL_API = "https://api.multimail.dev/v1"
HEADERS = {"Authorization": "Bearer mm_live_your_api_key"}

class EmailDraft(BaseModel):
    to: str
    subject: str
    body: str
    urgency: str = "normal"

    @field_validator("subject")
    @classmethod
    def subject_not_empty(cls, v):
        if not v.strip():
            raise ValueError("Subject cannot be empty")
        return v

    @field_validator("body")
    @classmethod
    def body_min_length(cls, v):
        if len(v) < 20:
            raise ValueError("Body must be at least 20 characters")
        return v

client = instructor.from_openai(OpenAI())

draft = client.chat.completions.create(
    model="gpt-4o",
    response_model=EmailDraft,
    messages=[{
        "role": "user",
        "content": "Send a follow-up to [email protected] about the proposal we discussed yesterday. Mention the $50k budget and Q3 timeline."
    }]
)
print(f"To: {draft.to}, Subject: {draft.subject}")

def send_via_multimail(draft: EmailDraft, mailbox_id: str) -> dict:
    """Send a validated EmailDraft through MultiMail."""
    resp = requests.post(f"{MULTIMAIL_API}/send", headers=HEADERS, json={
        "mailbox_id": mailbox_id,
        "to": draft.to,
        "subject": draft.subject,
        "body": draft.body
    })
    return resp.json()

"cm"># Extract and send in one flow
def compose_and_send(user_request: str, mailbox_id: str) -> dict:
    """Extract structured email from natural language and send via MultiMail."""
    draft = client.chat.completions.create(
        model="gpt-4o",
        response_model=EmailDraft,
        max_retries=3,  "cm"># Auto-retry on validation failure
        messages=[{
            "role": "system",
            "content": "Extract a professional email from the user&"cm">#039;s request. "
            "Include proper greeting and sign-off."
        }, {
            "role": "user",
            "content": user_request
        }]
    )
    # In gated_send mode, this queues for human approval
    return send_via_multimail(draft, mailbox_id)

result = compose_and_send(
    "Email the team at [email protected] that the deploy is pushed to Thursday",
    "your_mailbox_id"
)

from enum import Enum
from typing import Optional

class EmailCategory(str, Enum):
    SUPPORT = "support"
    SALES = "sales"
    BILLING = "billing"
    SPAM = "spam"
    OTHER = "other"

class EmailClassification(BaseModel):
    category: EmailCategory
    urgency: str
    summary: str
    suggested_action: str
    auto_reply_appropriate: bool

def classify_email(email_subject: str, email_body: str) -> EmailClassification:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=EmailClassification,
        messages=[{
            "role": "user",
            "content": f"Classify this email:\nSubject: {email_subject}\n\n{email_body}"
        }]
    )

"cm"># Fetch and classify inbox
resp = requests.get(
    f"{MULTIMAIL_API}/mailboxes/your_mailbox_id/inbox",
    headers=HEADERS, params={"limit": 10}
)
for email in resp.json().get("emails", []):
    classification = classify_email(email["subject"], email["body"])
    print(f"{email[&"cm">#039;subject']}: {classification.category} ({classification.urgency})")
    if classification.auto_reply_appropriate:
        "cm"># Generate and send auto-reply through MultiMail
        pass

from typing import Iterable

def extract_multiple_emails(request: str) -> list[EmailDraft]:
    """Extract multiple email drafts from a batch request."""
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=Iterable[EmailDraft],
        messages=[{
            "role": "system",
            "content": "Extract each individual email from the user&"cm">#039;s request. "
            "Each email should have a distinct recipient, subject, and body."
        }, {
            "role": "user",
            "content": request
        }]
    )

# Extract and send batch
drafts = extract_multiple_emails(
    "Send updates to the team: tell [email protected] the design is approved, "
    "tell [email protected] we need the API docs by Friday, "
    "and tell [email protected] the budget is confirmed."
)

for draft in drafts:
    result = send_via_multimail(draft, "your_mailbox_id")
    print(f"Sent to {draft.to}: {result}")