File: /home/clawbot-insurance/Skills/claims/claim-intake/claim_intake.py
#!/usr/bin/env python3"""Claim Intake SkillHandles First Notice of Loss (FNOL), collects initial information"""
import jsonimport refrom datetime import datetimefrom pathlib import Pathfrom typing import Dict, List
class ClaimIntake: # Claim types and required information CLAIM_TYPES = { "auto_collision": { "description": "Auto collision/collision claim", "required_info": [ "date_of_loss", "time_of_loss", "location", "vehicles_involved", "drivers", "injuries", "police_report", "witnesses", "photos" ], "questions": [ "When did the accident occur?", "Where did the accident happen?", "Were there any injuries?", "Was a police report filed?", "Were there any witnesses?", "Do you have photos of the damage?" ] }, "property_damage": { "description": "Property damage claim", "required_info": [ "date_of_loss", "cause_of_loss", "damaged_property", "extent_of_damage", "repair_estimates", "photos" ], "questions": [ "When did the damage occur?", "What caused the damage?", "What property was damaged?", "Do you have repair estimates?", "Do you have photos of the damage?" ] }, "liability": { "description": "Liability claim", "required_info": [ "date_of_incident", "claimant_info", "alleged_injuries", "witnesses", "police_report", "demand_letter" ], "questions": [ "When did the incident occur?", "Who is making the claim?", "What injuries are being claimed?", "Were there witnesses?", "Have you received a demand letter?" ] }, "workers_comp": { "description": "Workers compensation claim", "required_info": [ "date_of_injury", "employee_info", "injury_description", "medical_treatment", "witnesses", "employer_report" ], "questions": [ "When was the employee injured?", "How did the injury occur?", "What medical treatment was provided?", "Were there witnesses?", "Has the employer been notified?" ] } } def __init__(self, insurer_id: str): self.insurer_id = insurer_id self.memory_dir = Path(f"/home/clawbot-insurance/Memory") def start_intake(self, policy_number: str, report_method: str = "phone") -> Dict: """Start new claim intake""" # Look up policy policy = self._lookup_policy(policy_number) if not policy: return { "status": "policy_not_found", "message": "Could not locate policy. Please verify the policy number." } # Create claim record claim_id = f"CLM-{datetime.now().strftime('%Y%m%d-%H%M%S')}" claim = { "claim_id": claim_id, "policy_number": policy_number, "policyholder_name": policy.get("policyholder_name"), "reported_at": datetime.now().isoformat(), "report_method": report_method, "status": "intake_in_progress", "claim_type": None, "intake_data": {}, "documents": [] } # Save claim claim_dir = self.memory_dir / f"Claims/{claim_id}" claim_dir.mkdir(parents=True, exist_ok=True) with open(claim_dir / "metadata.json", 'w') as f: json.dump(claim, f, indent=2) return { "status": "intake_started", "claim_id": claim_id, "policyholder_name": policy.get("policyholder_name"), "next_question": "What type of claim are you reporting? (auto collision, property damage, liability, workers comp)" } def process_response(self, claim_id: str, field: str, value: str) -> Dict: """Process claimant response""" # Load claim claim_file = self.memory_dir / f"Claims/{claim_id}/metadata.json" with open(claim_file) as f: claim = json.load(f) # Store response claim["intake_data"][field] = value # If claim type just identified, load questions if field == "claim_type": claim_type = value.lower().replace(" ", "_") claim["claim_type"] = claim_type if claim_type in self.CLAIM_TYPES: claim["intake_questions"] = self.CLAIM_TYPES[claim_type]["questions"] claim["current_question_index"] = 0 # Get next question if claim.get("intake_questions"): current_idx = claim.get("current_question_index", 0) if current_idx < len(claim["intake_questions"]): next_question = claim["intake_questions"][current_idx] claim["current_question_index"] = current_idx + 1 # Save progress with open(claim_file, 'w') as f: json.dump(claim, f, indent=2) return { "status": "in_progress", "claim_id": claim_id, "next_question": next_question, "questions_remaining": len(claim["intake_questions"]) - current_idx - 1 } else: # Intake complete claim["status"] = "intake_complete" claim["intake_completed_at"] = datetime.now().isoformat() with open(claim_file, 'w') as f: json.dump(claim, f, indent=2) # Trigger next workflow return { "status": "intake_complete", "claim_id": claim_id, "message": "Thank you for providing this information. A claims adjuster will review your claim and contact you within 24 hours.", "next_steps": ["document_collection", "coverage_verification"] } # Save progress with open(claim_file, 'w') as f: json.dump(claim, f, indent=2) return { "status": "in_progress", "claim_id": claim_id } def _lookup_policy(self, policy_number: str) -> Dict: """Look up policy by number""" # Search policy index policies_dir = self.memory_dir / "Policies" for policy_dir in policies_dir.iterdir(): if policy_dir.is_dir(): policy_file = policy_dir / "policy-data.json" if policy_file.exists(): with open(policy_file) as f: policy = json.load(f) if policy.get("policy_number") == policy_number: return policy return None
# CLI interfaceif __name__ == "__main__": import sys insurer_id = sys.argv[1] command = sys.argv[2] intake = ClaimIntake(insurer_id) if command == "start": policy_number = sys.argv[3] report_method = sys.argv[4] if len(sys.argv) > 4 else "phone" result = intake.start_intake(policy_number, report_method) print(json.dumps(result, indent=2)) elif command == "respond": claim_id = sys.argv[3] field = sys.argv[4] value = sys.argv[5] result = intake.process_response(claim_id, field, value) print(json.dumps(result, indent=2))