Migrate PO processor to Bedrock

Switch the PO email processor's AI extraction from the Anthropic SDK to
bedrock-runtime InvokeModel on the inference profile
us.anthropic.claude-haiku-4-5-20251001-v1:0 (BEDROCK_MODEL_ID env), so
it no longer needs a provider API key or Secrets Manager secret. PO
parsing stays fully AI -- only the provider changes. The EXTRACTION_PROMPT
is kept byte-identical and the Bedrock text output is still decoded with
json.loads(..., parse_float=Decimal), which DynamoDB requires (it rejects
floats). Replace the deprecated datetime.utcnow() with
datetime.now(timezone.utc) and drop the anthropic dependency.
This commit is contained in:
Adam Moussa 2026-07-15 20:10:14 -04:00
parent 5e1993f9e4
commit 3151de92af
No known key found for this signature in database
2 changed files with 24 additions and 29 deletions

View file

@ -11,11 +11,10 @@ import json
import logging import logging
import os import os
import re import re
from datetime import datetime from datetime import datetime, timezone
from decimal import Decimal from decimal import Decimal
from email import policy from email import policy
import anthropic
import boto3 import boto3
logger = logging.getLogger() logger = logging.getLogger()
@ -23,9 +22,12 @@ logger.setLevel(logging.INFO)
s3 = boto3.client("s3") s3 = boto3.client("s3")
dynamodb = boto3.resource("dynamodb") dynamodb = boto3.resource("dynamodb")
bedrock = boto3.client("bedrock-runtime")
PO_TABLE = os.environ.get("PO_TABLE", "purchase-orders") PO_TABLE = os.environ.get("PO_TABLE", "purchase-orders")
ANTHROPIC_API_KEY_SECRET_ARN = os.environ.get("ANTHROPIC_API_KEY_SECRET_ARN") BEDROCK_MODEL_ID = os.environ.get(
"BEDROCK_MODEL_ID", "us.anthropic.claude-haiku-4-5-20251001-v1:0"
)
EXTRACTION_PROMPT = """\ EXTRACTION_PROMPT = """\
You are an email parser for a purchase order ingest pipeline. You are an email parser for a purchase order ingest pipeline.
@ -210,17 +212,6 @@ The Coupa commodity/category field if present in the email (e.g., "Maintenance -
""" """
def get_anthropic_client() -> anthropic.Anthropic:
"""Create Anthropic client, fetching API key from Secrets Manager."""
if ANTHROPIC_API_KEY_SECRET_ARN:
secrets = boto3.client("secretsmanager")
secret = secrets.get_secret_value(SecretId=ANTHROPIC_API_KEY_SECRET_ARN)
api_key = secret["SecretString"]
return anthropic.Anthropic(api_key=api_key)
# Fall back to ANTHROPIC_API_KEY env var (for local testing)
return anthropic.Anthropic()
def parse_raw_email(raw_bytes: bytes) -> dict: def parse_raw_email(raw_bytes: bytes) -> dict:
"""Parse a raw MIME email into subject, sender, and body text.""" """Parse a raw MIME email into subject, sender, and body text."""
msg = email.message_from_bytes(raw_bytes, policy=policy.default) msg = email.message_from_bytes(raw_bytes, policy=policy.default)
@ -252,9 +243,7 @@ def parse_raw_email(raw_bytes: bytes) -> dict:
def extract_with_claude(email_data: dict) -> dict: def extract_with_claude(email_data: dict) -> dict:
"""Send parsed email to Claude for structured extraction.""" """Send parsed email to Claude on Bedrock for structured extraction."""
client = get_anthropic_client()
email_text = ( email_text = (
f"Subject: {email_data['subject']}\n" f"Subject: {email_data['subject']}\n"
f"From: {email_data['sender']}\n" f"From: {email_data['sender']}\n"
@ -264,24 +253,29 @@ def extract_with_claude(email_data: dict) -> dict:
f"{email_data['body']}" f"{email_data['body']}"
) )
response = client.messages.create( resp = bedrock.invoke_model(
model="claude-haiku-4-5-20251001", modelId=BEDROCK_MODEL_ID,
max_tokens=2048, body=json.dumps(
messages=[
{ {
"role": "user", "anthropic_version": "bedrock-2023-05-31",
"content": f"{EXTRACTION_PROMPT}\n\nEMAIL:\n{email_text}", "max_tokens": 2048,
"messages": [
{
"role": "user",
"content": f"{EXTRACTION_PROMPT}\n\nEMAIL:\n{email_text}",
}
],
} }
], ),
) )
response_text = json.loads(resp["body"].read())["content"][0]["text"]
response_text = response.content[0].text
# Extract JSON from response (handle markdown code blocks) # Extract JSON from response (handle markdown code blocks)
json_match = re.search(r"```(?:json)?\s*(.*?)```", response_text, re.DOTALL) json_match = re.search(r"```(?:json)?\s*(.*?)```", response_text, re.DOTALL)
if json_match: if json_match:
response_text = json_match.group(1) response_text = json_match.group(1)
# parse_float=Decimal is CRITICAL: DynamoDB rejects Python floats.
return json.loads(response_text.strip(), parse_float=Decimal) return json.loads(response_text.strip(), parse_float=Decimal)
@ -296,7 +290,7 @@ def pad_zip(zip_code: str | None) -> str | None:
def enrich_parsed(parsed: dict, s3_key: str, email_subject: str): def enrich_parsed(parsed: dict, s3_key: str, email_subject: str):
"""Add metadata and promote nested fields to top level.""" """Add metadata and promote nested fields to top level."""
now = datetime.utcnow().isoformat() now = datetime.now(timezone.utc).isoformat()
parsed["raw_s3_key"] = s3_key parsed["raw_s3_key"] = s3_key
parsed["processed_at"] = now parsed["processed_at"] = now
parsed["data_source"] = "email" parsed["data_source"] = "email"
@ -346,7 +340,9 @@ def save_cancellation(parsed: dict):
UpdateExpression="SET po_status = :status, cancelled_at = :cancelled_at, raw_s3_key = :s3_key", UpdateExpression="SET po_status = :status, cancelled_at = :cancelled_at, raw_s3_key = :s3_key",
ExpressionAttributeValues={ ExpressionAttributeValues={
":status": "Cancelled", ":status": "Cancelled",
":cancelled_at": parsed.get("processed_at", datetime.utcnow().isoformat()), ":cancelled_at": parsed.get(
"processed_at", datetime.now(timezone.utc).isoformat()
),
":s3_key": parsed.get("raw_s3_key", ""), ":s3_key": parsed.get("raw_s3_key", ""),
}, },
) )

View file

@ -1,2 +1 @@
anthropic>=0.116.0
boto3>=1.43.47 boto3>=1.43.47