import logging from typing import Dict, Any, Optional, List, Callable # Import lead_scoring_agent for lead scoring integration from lead_scoring_agent import score_lead from data_enrichment import DataEnrichmentManager class OutreachAgent: """ OutreachAgent handles lead intake, message tailoring, and multi-channel delivery. Integrates lead_scoring_agent to score leads at intake for prioritization and routing. """ def __init__(self, username: str, channel_handlers: Optional[Dict[str, Callable]]=None): self.logger = logging.getLogger("OutreachAgent") self.lead_assets = {} self.analytics = [] self.channel_handlers = channel_handlers or {} self.username = username self.enrichment_manager = DataEnrichmentManager(username) def intake_lead(self, lead_profile: Dict[str, Any]) -> Dict[str, Any]: """ Intake lead with scoring integration as first logic step. Calls score_lead(profile) to enrich profile with scoring metadata. """ # FIRST STEP: Score the lead using lead_scoring_agent scored_profile = score_lead(lead_profile) self.logger.info(f"Lead scored: {scored_profile.get('name', 'Unknown')} - Score: {scored_profile.get('lead_score', 'N/A')}") # Enrich profile with additional data enriched_profile = self.validate_and_enrich_profile(scored_profile) self.logger.info(f"Intake for {enriched_profile.get('name', 'Unknown Lead')}") asset_id = enriched_profile.get('id') if asset_id: self.lead_assets[asset_id] = enriched_profile return enriched_profile def validate_and_enrich_profile(self, profile: Dict[str, Any]) -> Dict[str, Any]: # Enrich with historical, persona, engagements, etc. # Preserve scoring metadata from lead_scoring_agent # ENRICHMENT STEP: Use DataEnrichmentManager to enrich profile with 14 APIs try: enriched_data = self.enrichment_manager.enrich_contact_full(profile) if enriched_data: # Merge enriched data into profile (preserve existing fields) for key, value in enriched_data.items(): if key not in profile or not profile[key]: # Only update if field is missing profile[key] = value self.logger.info(f"Profile enriched with {len(enriched_data)} data sources") except Exception as e: self.logger.warning(f"Enrichment failed: {e}. Continuing with available data.") required_fields = ['persona', 'channel_fit'] for field in required_fields: if field not in profile: profile[field] = 'unspecified' # Simulate enrichment logic profile['validation'] = 'success' if 'recent_posts' not in profile: profile['recent_posts'] = [] return profile def select_template(self, profile: Dict[str, Any]) -> str: persona = profile.get('persona', 'Generic') pain = profile.get('pain_opportunity', 'N/A') buyer_signals = profile.get('buyer_signals', []) channel = profile.get('channel_fit', 'email') # Template selection logic (expandable) template = f"Hi {{name}}, as a {{persona}}, we noticed: {{pain}}. {{buyer_signals}}. Ready to discuss via {{channel}}?" return template def tailor_message(self, profile: Dict[str, Any], template: str) -> str: variables = { "name": profile.get("name", "there"), "persona": profile.get("persona", "Professional"), "pain": profile.get("pain_opportunity", "key considerations"), "buyer_signals": ', '.join(profile.get("buyer_signals", [])), "channel": profile.get("channel_fit", "email"), "context": profile.get("context_snippets", ""), "recent_posts": '\n'.join(profile.get("recent_posts", [])), "complaints": ', '.join(profile.get("customer_complaint_tags", [])), "hooks": ', '.join(profile.get("competitor_industry_hooks", [])), "urgency": profile.get("urgency_triggers", "") } message = template.format(**variables) # Attach recent posts, urgency, hooks etc with nuance if variables["context"]: message += f"\nContext: {variables['context']}" if variables["recent_posts"]: message += f"\nRecent posts: {variables['recent_posts']}" if variables["complaints"]: message += f"\nCustomer complaints: {variables['complaints']}" if variables["hooks"]: message += f"\nIndustry/competitor hooks: {variables['hooks']}" if variables["urgency"]: message += f"\nUrgency triggers: {variables['urgency']}" return message def deliver(self, profile: Dict[str, Any], message: str, channel: str) -> Dict[str, Any]: # Channel handler pattern for extensibility handler = self.channel_handlers.get(channel, self.default_handler) result = handler(profile, message) self.logger.info(f"Delivered to {profile.get('name')} via {channel}") self.update_analytics(profile, channel, result) return result def default_handler(self, profile: Dict[str, Any], message: str) -> Dict[str, Any]: # Simulate sending logic return {"status": "sent", "channel": profile.get('channel_fit', 'email'), "message": message} def browser_social_actions(self, profile: Dict[str, Any], action: str, extra: Optional[Dict[str, Any]]=None) -> Dict[str, Any]: # Simulate browser agent: connect, DM, comment, like, milestone outreach valid_actions = ["connect", "dm", "comment", "like", "milestone"] if action not in valid_actions: raise ValueError(f"Unsupported social action: {action}") self.logger.info(f"Performed {action} for {profile.get('name')}") result = {"action": action, "status": "success", "lead": profile.get('id')} self.update_analytics(profile, "social_action", result) return result def update_analytics(self, profile: Dict[str, Any], channel: str, result: Dict[str, Any]): entry = { "lead_id": profile.get("id"), "channel": channel, "result": result } self.analytics.append(entry) if profile.get("id") in self.lead_assets: self.lead_assets[profile.get("id")]["last_engagement"] = entry def feedback_loop(self, profile: Dict[str, Any], feedback: str): # Store feedback and auto-update asset pid = profile.get("id") if pid and pid in self.lead_assets: self.lead_assets[pid]["feedback"] = feedback self.logger.info(f"Feedback for {profile.get('name')}: {feedback}") def get_transcript(self, profile_id: str) -> Dict[str, Any]: asset = self.lead_assets.get(profile_id) return { "profile": asset, "last_message": asset.get("last_engagement", {}).get("result", {}).get("message", "") if asset else "", "engagement_stats": [a for a in self.analytics if a.get("lead_id") == profile_id] } # Example usage handler for email def email_handler(profile, message): # Integrate real email API here return {"status": "sent_via_email", "recipient": profile.get("email", "n/a"), "message": message} # Example handlers for expansion for LinkedIn, SMS, browser agent, etc. def linkedin_handler(profile, message): # Integrate real LinkedIn API calls here return {"status": "sent_via_linkedin", "recipient": profile.get("linkedin", "n/a"), "message": message} def sms_handler(profile, message): # Integrate real SMS API calls here return {"status": "sent_via_sms", "recipient": profile.get("phone", "n/a"), "message": message} # Usage: # agent = OutreachAgent(channel_handlers={"email": email_handler, "linkedin": linkedin_handler, "sms": sms_handler}) # lead = agent.intake_lead({...}) # template = agent.select_template(lead) # personalized_msg = agent.tailor_message(lead, template) # agent.deliver(lead, personalized_msg, channel="email") # agent.browser_social_actions(lead, "connect")