"""
def get_client_status_html() -> str:
if knowledge_base["client"]["name"]:
return f"""
β
Client Profile Active
AI is finding prospects for {knowledge_base["client"]["name"]}
"""
return """
β οΈ
Setup Required
Go to Setup tab to enter your company name and start AI prospect discovery.
"""
def get_dashboard_stats():
return (
get_stat_html(str(len(knowledge_base["prospects"])), "Prospects Found", "var(--primary-blue)"),
get_stat_html(str(len(knowledge_base["contacts"])), "Decision Makers", "var(--success-green)"),
get_stat_html(str(len(knowledge_base["emails"])), "Emails Drafted", "var(--warning-orange)"),
get_client_status_html()
)
def merge_to_knowledge_base(prospects_found: list, contacts_found: list, emails_drafted: list):
"""Merge found data to knowledge base with deduplication"""
global knowledge_base
# Deduplicate prospects by name/domain
existing_prospect_keys = set()
for p in knowledge_base["prospects"]:
key = (p.get("name", "").lower(), p.get("domain", "").lower())
existing_prospect_keys.add(key)
for p in prospects_found:
key = (p.get("name", "").lower(), p.get("domain", "").lower())
if key not in existing_prospect_keys:
knowledge_base["prospects"].append(p)
existing_prospect_keys.add(key)
# Deduplicate contacts by email
existing_emails = set(c.get("email", "").lower() for c in knowledge_base["contacts"])
for c in contacts_found:
email = c.get("email", "").lower()
if email and email not in existing_emails:
knowledge_base["contacts"].append(c)
existing_emails.add(email)
# Deduplicate emails by to+subject
existing_email_keys = set()
for e in knowledge_base["emails"]:
key = (e.get("to", "").lower(), e.get("subject", "").lower())
existing_email_keys.add(key)
for e in emails_drafted:
key = (e.get("to", "").lower(), e.get("subject", "").lower())
if key not in existing_email_keys:
knowledge_base["emails"].append(e)
existing_email_keys.add(key)
def get_prospects_html() -> str:
if not knowledge_base["prospects"]:
return """
π―
No prospects discovered yet
Complete the Setup and click "Find Prospects" to let AI discover potential customers
"""
html = ""
for p in reversed(knowledge_base["prospects"]):
status_class = "badge-researched" if p.get("research_complete") else "badge-new"
status_text = "RESEARCHED" if p.get("research_complete") else "DISCOVERED"
# Build contacts list (case-insensitive matching)
contacts_html = ""
p_name_lower = p.get("name", "").lower()
prospect_contacts = [c for c in knowledge_base["contacts"]
if p_name_lower in c.get("company", "").lower()
or c.get("company", "").lower() in p_name_lower]
if prospect_contacts:
contacts_html = "
{'β Email drafted' if p.get("email_drafted") else 'β³ Pending'}
π Discovered
{p.get("discovered_at") or datetime.now().strftime("%Y-%m-%d %H:%M")}
"""
return html
def get_emails_html() -> str:
if not knowledge_base["emails"]:
return """
βοΈ
No emails drafted yet
AI will draft personalized emails after discovering prospects
"""
html = ""
for e in reversed(knowledge_base["emails"]):
body_display = e.get("body", "").replace("\n", " ")
html += f"""
βοΈ {e.get("subject", "No subject")[:50]}{'...' if len(e.get("subject", "")) > 50 else ''}DRAFT
π’ Prospect
{e.get("prospect_company", "Unknown")}
π§ To
{e.get("to", "Not specified")}
π Subject
{e.get("subject", "No subject")}
π Email Body
{body_display}
"""
return html
def get_contacts_html() -> str:
if not knowledge_base["contacts"]:
return """
π₯
No contacts found yet
AI will find decision makers when discovering prospects
"""
html = """
β Verified Contacts: All contacts shown here were found through web searches of LinkedIn profiles,
company team pages, and public directories. Only contacts with verified email addresses found on the web are displayed.
"""
for c in reversed(knowledge_base["contacts"]):
source = c.get("source", "web_search")
source_label = {
"web_search": "Found via web search",
"linkedin": "Found via LinkedIn",
"team_page": "Found on company page",
"web_search_and_scraping": "Verified from web"
}.get(source, "Verified")
html += f"""
π€ {c.get("name", "Unknown")}
{c.get("title", "Unknown title")}
π’ {c.get("company", "Unknown company")}
{f'
π§ {c.get("email")}
' if c.get("email") else ''}
VERIFIED
{source_label}
"""
return html
def reset_all_data():
global knowledge_base
knowledge_base = {
"client": {"name": None, "industry": None, "target_market": None, "products_services": None,
"value_proposition": None, "ideal_customer_profile": None, "researched_at": None, "raw_research": None},
"prospects": [], "contacts": [], "emails": [], "chat_history": []
}
stats = get_dashboard_stats()
return (stats[0], stats[1], stats[2], stats[3], get_prospects_html(), get_emails_html(),
get_contacts_html(), "", "*Enter your company name to begin.*", "*Click 'Find Prospects' after setup.*")
# ============================================================================
# CLIENT SETUP - Research the user's company
# ============================================================================
async def setup_client_company(company_name: str, hf_token_input: str, serper_key_input: str = "", progress=gr.Progress()):
global knowledge_base
if not company_name or not company_name.strip():
yield "β οΈ Please enter your company name."
return
# Get HF token from UI input or environment
token = get_hf_token(hf_token_input)
if not token:
yield "β οΈ **HF_TOKEN Required**: Please enter your HuggingFace token in the Setup tab.\n\nGet a free token at: https://huggingface.co/settings/tokens"
return
# Store SERPER API key if provided (prioritize user input)
if serper_key_input and serper_key_input.strip():
get_serper_key(serper_key_input)
# Update the search service with current key
update_search_service_key()
company_name = company_name.strip()
# Initialize progress log with HTML styling
output = f"""
"""
yield output
progress(0.2)
except Exception as e:
yield f"""
βAgent init failed: {e}
"""
return
task = f"""Research {company_name} to understand their business. Use search_web to find information about:
1. What {company_name} does - their products/services
2. Their target market and ideal customers
3. Their industry and market position
4. Their value proposition
5. What type of companies would be good prospects for them
Use the save_company tool to save information about {company_name}:
- company_id: "{company_name.lower().replace(' ', '_')}"
- name: "{company_name}"
- domain: their website domain
- industry: their industry
- description: brief company description
After researching, provide a comprehensive summary of:
- What {company_name} does
- Who their ideal customers are
- What industries/company types would benefit from their services
This is OUR company - we need this information to find matching prospects."""
last_research = "" # Track last AI response for fallback
search_results_summary = [] # Capture actual search results
search_count = 0
try:
async for event in agent.run(task, max_iterations=12):
event_type = event.get("type")
if event_type == "model_loaded":
output += f"""
"""
elif tool in ["save_company", "save_fact"]:
output += f"""
πΎSaving information...
"""
yield output
progress(0.3 + min(search_count * 0.1, 0.4))
elif event_type == "tool_result":
tool = event.get("tool", "")
result = event.get("result", {})
if tool in ["search_web", "search_news"]:
count = result.get("count", 0) if isinstance(result, dict) else 0
output += f"""
β Found {count} results
"""
# Capture search results for building a summary
if isinstance(result, dict) and result.get("results"):
for r in result.get("results", [])[:3]: # Top 3 results
if isinstance(r, dict):
title = r.get("title", "")
# Try multiple field names for snippet/body
snippet = r.get("body", r.get("text", r.get("snippet", r.get("description", ""))))
if title and title not in str(search_results_summary):
if snippet:
search_results_summary.append(f"- **{title}**: {snippet[:200]}..." if len(snippet) > 200 else f"- **{title}**: {snippet}")
else:
search_results_summary.append(f"- **{title}**")
yield output
elif event_type == "thought":
# Capture AI thoughts for potential use as research summary
thought = event.get("thought", "")
message = event.get("message", "")
# Filter out any HTML/footer content that might leak through
if thought and not thought.startswith("CX AI Agent") and "Powered by AI" not in thought and not thought.startswith("[Processing:"):
if len(thought) > len(last_research):
last_research = thought
logger.info(f"Captured research thought: {thought[:100]}...")
# Also show progress in output
output += f"π {message}\n"
yield output
elif message:
# Show reasoning progress even if thought is minimal
output += f"π€ {message}\n"
yield output
elif event_type == "agent_complete":
final_answer = event.get("final_answer", "")
# Filter out HTML footer that might leak through
if not final_answer or "CX AI Agent" in final_answer or "Powered by AI" in final_answer:
final_answer = last_research
# If still no answer, build from search results
if not final_answer and search_results_summary:
final_answer = f"**{company_name}** - Research findings:\n\n" + "\n".join(search_results_summary[:10])
if not final_answer:
final_answer = f"Research completed for {company_name}. The AI gathered information about the company. Ready to find prospects."
knowledge_base["client"] = {
"name": company_name,
"raw_research": final_answer,
"researched_at": datetime.now().strftime("%Y-%m-%d %H:%M")
}
output += f"\n---\n\n## β {company_name} Profile Complete!\n\n"
output += "**Next step:** Go to the **Discovery** tab and click **'π Find Prospects & Contacts'** to let AI discover potential customers.\n\n"
# Show search results if we have them
if search_results_summary:
output += "---\n\n### π Search Results Found\n\n"
output += "\n".join(search_results_summary[:8])
output += "\n\n"
output += f"---\n\n### π Research Summary\n\n{final_answer}"
yield output
progress(1.0)
return
elif event_type == "agent_max_iterations":
# Still save what we have
final_answer = last_research
if not final_answer and search_results_summary:
final_answer = f"**{company_name}** - Research findings:\n\n" + "\n".join(search_results_summary[:10])
if not final_answer:
final_answer = f"Research completed for {company_name}. Ready to find prospects."
knowledge_base["client"] = {
"name": company_name,
"raw_research": final_answer,
"researched_at": datetime.now().strftime("%Y-%m-%d %H:%M")
}
output += f"\n---\n\n## β {company_name} Profile Complete!\n\n"
output += "**Next step:** Go to the **Discovery** tab and click **'π Find Prospects & Contacts'** to let AI discover potential customers.\n\n"
if final_answer:
output += f"---\n\n### π Research Summary\n\n{final_answer}"
yield output
progress(1.0)
return
elif event_type == "agent_error":
error_msg = event.get("error", "Unknown error")
# Still save basic profile so user can proceed
knowledge_base["client"] = {
"name": company_name,
"raw_research": last_research or f"{company_name} - manual research may be needed.",
"researched_at": datetime.now().strftime("%Y-%m-%d %H:%M")
}
output += f"\nβ οΈ AI encountered an issue: {error_msg}\n"
output += f"\n---\n\n## β οΈ {company_name} Setup (Partial)\n\n"
output += "**Note:** Some research may be incomplete. You can still proceed to find prospects.\n\n"
yield output
progress(1.0)
return
except Exception as e:
# Save basic profile on exception so user can still proceed
knowledge_base["client"] = {
"name": company_name,
"raw_research": last_research or f"{company_name} - setup interrupted.",
"researched_at": datetime.now().strftime("%Y-%m-%d %H:%M")
}
output += f"\nβ οΈ Error: {e}\n"
output += f"\n**Note:** Basic profile saved. You can still try to find prospects.\n"
yield output
return
# If we get here without returning, the loop completed without agent_complete/max_iterations/error
# This means the agent just stopped - save what we have
if not knowledge_base["client"]["name"]:
final_answer = last_research
if not final_answer and search_results_summary:
final_answer = f"**{company_name}** - Research findings:\n\n" + "\n".join(search_results_summary[:10])
if not final_answer:
final_answer = f"Research completed for {company_name}. Ready to find prospects."
knowledge_base["client"] = {
"name": company_name,
"raw_research": final_answer,
"researched_at": datetime.now().strftime("%Y-%m-%d %H:%M")
}
output += f"\n---\n\n## β {company_name} Profile Complete!\n\n"
output += "**Next step:** Go to the **Discovery** tab and click **'π Find Prospects & Contacts'** to let AI discover potential customers.\n\n"
output += f"---\n\n### π Research Summary\n\n{final_answer}"
yield output
# ============================================================================
# AI PROSPECT DISCOVERY - Automatically find prospects
# ============================================================================
async def discover_prospects(num_prospects: int, progress=gr.Progress()):
global knowledge_base
if not knowledge_base["client"]["name"]:
yield "β οΈ **Setup Required**: Please go to Setup tab and enter your company name first."
return
# Use session token (set in Setup tab)
token = session_hf_token.get("token")
if not token:
yield "β οΈ **HF_TOKEN Required**: Please enter your HuggingFace token in the **Setup** tab first.\n\nGet a free token at: https://huggingface.co/settings/tokens"
return
# Ensure search service has current SERPER key
update_search_service_key()
client_name = knowledge_base["client"]["name"]
client_info = knowledge_base["client"].get("raw_research", "")
# Initialize progress log with collapsible accordion
progress_steps = []
def build_accordion(steps, is_loading=True, summary_html=""):
"""Build the collapsible accordion HTML"""
status_text = "Processing..." if is_loading else "Complete"
spinner = '' if is_loading else 'β '
steps_html = ""
for step in steps:
icon_class = step.get("icon_class", "tool")
steps_html += f'''
{step.get("icon", "π§")}
{step.get("title", "")}
{f'
{step.get("detail", "")}
' if step.get("detail") else ""}
'''
return f'''
{spinner}
π AI Discovery Progress - {status_text}
βΌ
{steps_html}
{summary_html}'''
progress_steps.append({"icon": "β³", "icon_class": "loading", "title": "Initializing AI agent...", "detail": f"Preparing to find prospects for {client_name}"})
yield build_accordion(progress_steps)
progress(0.1)
try:
# Initialize HuggingFace agent with nscale provider
agent = AutonomousMCPAgentHF(
mcp_registry=mcp_registry,
hf_token=token,
provider=HF_PROVIDER,
model=HF_MODEL
)
progress_steps[-1] = {"icon": "β ", "icon_class": "success", "title": "AI Agent initialized", "detail": f"Model: {agent.model}"}
yield build_accordion(progress_steps)
progress(0.2)
except Exception as e:
progress_steps[-1] = {"icon": "β", "icon_class": "error", "title": "Agent initialization failed", "detail": str(e)[:100]}
yield build_accordion(progress_steps, is_loading=False)
return
# Build a concise industry description from client research
# This helps the discovery tool generate better search queries
client_industry_desc = f"{client_name}"
if client_info:
# Extract key info - first 200 chars or first sentence
info_snippet = client_info[:300].split('.')[0] if '.' in client_info[:300] else client_info[:200]
client_industry_desc = f"{client_name} - {info_snippet}"
task = f"""You are an AI sales agent finding prospects for {client_name}.
About {client_name}:
{client_info}
USE THE discover_prospects_with_contacts TOOL - it handles everything automatically:
- Searches for potential prospect companies (CUSTOMERS who would buy from {client_name})
- Finds verified contacts for each (LinkedIn, company websites, directories, etc.)
- ONLY saves prospects that have real verified contacts
- Keeps searching until target is met or max attempts reached
- Skips companies without contacts automatically
STEP 1: Call discover_prospects_with_contacts with accurate industry description:
{{"client_company": "{client_name}", "client_industry": "{client_industry_desc}", "target_prospects": {num_prospects}, "target_titles": ["CEO", "Founder", "VP Sales", "CTO", "Head of Sales"]}}
STEP 2: After discovery completes, for each prospect with contacts, draft personalized email:
- Use send_email tool with the REAL contact info returned
- to: actual verified email
- subject: Reference {client_name} AND the prospect's business
- body: Personalized email mentioning the contact by name and specific facts about their company
- prospect_id: the prospect_id from discovery results
IMPORTANT:
- The discover_prospects_with_contacts tool does ALL the hard work
- It will check multiple companies until it finds {num_prospects} with verified contacts
- Only prospects WITH contacts are saved (no useless data)
- NEVER invent contact names or emails - only use what the tool returns
After the tool completes, provide a summary of:
- Prospects saved (with verified contacts)
- Total contacts found
- Companies checked vs skipped
- Emails drafted"""
prospects_found = []
contacts_found = []
emails_drafted = []
search_results_for_prospects = [] # Capture search results to extract prospects
# Track pending tool calls to capture data
pending_prospect = None
pending_contact = None
current_prospect_name = None # Track which prospect we're working on
try:
iteration = 0
last_final_answer = "" # Track the last complete response from AI
async for event in agent.run(task, max_iterations=25):
event_type = event.get("type")
iteration += 1
progress_pct = min(0.2 + (iteration * 0.03), 0.95)
if event_type == "model_loaded":
progress_steps.append({"icon": "π§ ", "icon_class": "success", "title": event.get('message', 'Model loaded'), "detail": ""})
yield build_accordion(progress_steps)
elif event_type == "iteration_start":
progress_steps.append({"icon": "π", "icon_class": "loading", "title": "AI is thinking...", "detail": event.get('message', '')})
yield build_accordion(progress_steps)
elif event_type == "tool_call":
tool = event.get("tool", "")
tool_input = event.get("input", {})
if tool == "search_web":
query = tool_input.get("query", "") if isinstance(tool_input, dict) else ""
progress_steps.append({
"icon": "π",
"icon_class": "tool",
"title": f'MCP search_web',
"detail": f'Query: "{query[:60]}{"..." if len(query) > 60 else ""}"'
})
elif tool == "search_news":
progress_steps.append({
"icon": "π°",
"icon_class": "tool",
"title": f'MCP search_news',
"detail": "Searching for recent news..."
})
elif tool == "discover_prospects_with_contacts":
target = tool_input.get("target_prospects", num_prospects) if isinstance(tool_input, dict) else num_prospects
progress_steps.append({
"icon": "π",
"icon_class": "tool",
"title": f'MCP discover_prospects_with_contacts',
"detail": f"Finding {target} prospects with verified contacts..."
})
elif tool == "save_prospect":
if isinstance(tool_input, dict):
company = tool_input.get("company_name", "Unknown")
current_prospect_name = company # Track current prospect
progress_steps.append({
"icon": "π―",
"icon_class": "success",
"title": f"Found prospect: {company}",
"detail": tool_input.get("company_domain", "")
})
# Capture prospect data during tool_call
pending_prospect = {
"name": company,
"domain": tool_input.get("company_domain", ""),
"summary": tool_input.get("metadata", {}).get("summary", "") if isinstance(tool_input.get("metadata"), dict) else "",
"industry": tool_input.get("metadata", {}).get("industry", "") if isinstance(tool_input.get("metadata"), dict) else "",
"fit_reason": tool_input.get("metadata", {}).get("fit_reason", "") if isinstance(tool_input.get("metadata"), dict) else "",
"fit_score": tool_input.get("fit_score", 0),
"research_complete": True,
"email_drafted": False,
"discovered_at": datetime.now().strftime("%Y-%m-%d %H:%M")
}
elif tool == "save_contact":
if isinstance(tool_input, dict):
# Handle both "name" and "first_name/last_name" formats
first_name = tool_input.get("first_name", "")
last_name = tool_input.get("last_name", "")
if first_name or last_name:
name = f"{first_name} {last_name}".strip()
else:
name = tool_input.get("name", "Unknown")
title = tool_input.get("title", "")
# Get company name - prioritize actual name over ID
company = tool_input.get("company_name") or current_prospect_name or "Unknown"
if company.startswith("company_") or company.startswith("prospect_"):
company = current_prospect_name or company
progress_steps.append({
"icon": "π€",
"icon_class": "success",
"title": f"Found contact: {name}",
"detail": f"{title} at {company}"
})
# Capture contact data during tool_call
pending_contact = {
"name": name,
"title": title or "Unknown",
"email": tool_input.get("email", ""),
"company": company,
"linkedin": tool_input.get("linkedin_url", "")
}
elif tool == "send_email":
progress_steps.append({
"icon": "βοΈ",
"icon_class": "tool",
"title": f'MCP send_email',
"detail": f"Drafting email for {current_prospect_name or 'prospect'}..."
})
if isinstance(tool_input, dict):
emails_drafted.append({
"to": tool_input.get("to", ""),
"subject": tool_input.get("subject", ""),
"body": tool_input.get("body", ""),
"prospect_company": current_prospect_name or tool_input.get("prospect_id", "Unknown"),
"created_at": datetime.now().strftime("%Y-%m-%d %H:%M")
})
elif tool == "find_verified_contacts":
company = tool_input.get("company_name", "company") if isinstance(tool_input, dict) else "company"
progress_steps.append({
"icon": "π",
"icon_class": "tool",
"title": f'MCP find_verified_contacts',
"detail": f"Looking for decision makers at {company}..."
})
yield build_accordion(progress_steps)
progress(progress_pct)
elif event_type == "tool_result":
tool = event.get("tool", "")
result = event.get("result", {})
if tool == "save_prospect":
if pending_prospect:
prospects_found.append(pending_prospect)
pending_prospect = None
elif tool == "save_contact":
if pending_contact:
contacts_found.append(pending_contact)
pending_contact = None
elif tool == "discover_prospects_with_contacts":
# Handle the all-in-one prospect discovery tool
if isinstance(result, dict):
status = result.get("status", "")
discovered_prospects = result.get("prospects", [])
total_contacts = result.get("contacts_count", 0)
companies_checked = result.get("companies_checked", 0)
companies_skipped = result.get("companies_skipped", 0)
message = result.get("message", "")
progress_steps.append({
"icon": "π",
"icon_class": "success",
"title": "Discovery Complete!",
"detail": f"Checked {companies_checked} companies, found {len(discovered_prospects)} with contacts"
})
if discovered_prospects:
for p in discovered_prospects:
# Add to prospects_found with complete data
prospect_data = {
"name": p.get("company_name", "Unknown"),
"domain": p.get("domain", ""),
"fit_score": p.get("fit_score", 75),
"summary": p.get("summary", f"Found with {p.get('contact_count', 0)} verified contacts"),
"industry": p.get("industry", "Technology & Services"),
"fit_reason": p.get("fit_reason", "Matches target customer profile based on industry and company size"),
"research_complete": True,
"email_drafted": False,
"discovered_at": datetime.now().strftime("%Y-%m-%d %H:%M")
}
prospects_found.append(prospect_data)
progress_steps.append({
"icon": "β ",
"icon_class": "success",
"title": f"{p.get('company_name')}",
"detail": f"{p.get('domain')} - {p.get('contact_count', 0)} contacts"
})
# Add contacts
for c in p.get("contacts", []):
contact_data = {
"name": c.get("name", "Unknown"),
"email": c.get("email", ""),
"title": c.get("title", ""),
"company": p.get("company_name", ""),
"verified": True,
"source": c.get("source", "web_search")
}
contacts_found.append(contact_data)
else:
progress_steps.append({
"icon": "β οΈ",
"icon_class": "warning",
"title": "No prospects with verified contacts found",
"detail": message
})
yield build_accordion(progress_steps)
elif tool == "find_verified_contacts":
# Handle verified contacts from the enhanced contact finder (single company)
if isinstance(result, dict):
status = result.get("status", "")
found_contacts = result.get("contacts", [])
message = result.get("message", "")
if status == "success" and found_contacts:
progress_steps.append({
"icon": "β ",
"icon_class": "success",
"title": f"Found {len(found_contacts)} verified contacts",
"detail": ", ".join([c.get("name", "") for c in found_contacts[:3]])
})
for c in found_contacts:
contact_data = {
"name": c.get("name", "Unknown"),
"email": c.get("email", ""),
"title": c.get("title", ""),
"company": c.get("company", current_prospect_name or ""),
"verified": c.get("verified", True),
"source": c.get("source", "web_search")
}
contacts_found.append(contact_data)
elif status == "no_contacts_found":
progress_steps.append({
"icon": "βοΈ",
"icon_class": "warning",
"title": "No contacts found",
"detail": message
})
yield build_accordion(progress_steps)
elif tool == "send_email":
progress_steps.append({
"icon": "β ",
"icon_class": "success",
"title": "Email drafted",
"detail": f"For {current_prospect_name or 'prospect'}"
})
# Mark prospect as having email drafted
if prospects_found:
prospects_found[-1]["email_drafted"] = True
yield build_accordion(progress_steps)
elif tool in ["search_web", "search_news"]:
count = result.get("count", 0) if isinstance(result, dict) else 0
# Update the last progress step with result count
if progress_steps and "search" in progress_steps[-1].get("title", "").lower():
progress_steps[-1]["detail"] += f" β Found {count} results"
# Capture search results to potentially extract prospects from
if isinstance(result, dict) and result.get("results"):
for r in result.get("results", []):
if isinstance(r, dict):
title = r.get("title", "")
snippet = r.get("body", r.get("text", r.get("snippet", r.get("description", ""))))
url = r.get("url", r.get("source", r.get("link", "")))
if title:
search_results_for_prospects.append({
"title": title,
"snippet": snippet,
"url": url
})
yield build_accordion(progress_steps)
elif event_type == "thought":
# Capture AI thoughts/responses as potential final answer
thought = event.get("thought", "")
message = event.get("message", "")
# Filter out HTML/garbage content
if thought and "CX AI Agent" not in thought and "Powered by AI" not in thought and not thought.startswith("[Processing:"):
last_final_answer = thought
elif event_type == "agent_complete":
# Auto-generate emails if AI didn't draft any but we have contacts
if contacts_found and not emails_drafted:
progress_steps.append({
"icon": "βοΈ",
"icon_class": "tool",
"title": "Auto-drafting outreach emails...",
"detail": f"Creating personalized emails for {len(contacts_found)} contacts"
})
yield build_accordion(progress_steps)
for c in contacts_found:
if c.get("email"):
contact_name = c.get("name", "").split()[0] if c.get("name") else "there"
full_name = c.get("name", "")
company = c.get("company", "your company")
title = c.get("title", "")
email_body = f"""Hi {contact_name},
I hope this message finds you well. I recently came across {company} and was genuinely impressed by the innovative work your team is doing in the industry.
As {title} at {company}, you're likely focused on driving growth and staying ahead of industry trends. That's exactly why I wanted to reach out.
At {client_name}, we specialize in helping companies like {company} achieve their strategic objectives through tailored solutions. We've helped similar organizations:
β’ Streamline their operations and reduce costs
β’ Accelerate growth through innovative strategies
β’ Stay competitive in an evolving market
I'd love to share some specific insights that have worked well for companies in your space. Would you be open to a brief 15-minute call this week to explore if there might be a fit?
I'm flexible on timing and happy to work around your schedule.
Looking forward to connecting,
Best regards,
{client_name} Team
P.S. If you're not the right person to speak with about this, I'd greatly appreciate it if you could point me in the right direction."""
emails_drafted.append({
"to": c.get("email"),
"subject": f"{contact_name}, quick question about {company}'s 2025 growth plans",
"body": email_body,
"prospect_company": company,
"contact_name": full_name,
"created_at": datetime.now().strftime("%Y-%m-%d %H:%M")
})
progress_steps.append({
"icon": "β ",
"icon_class": "success",
"title": f"Drafted {len(emails_drafted)} outreach emails",
"detail": "Ready for review in the Emails tab"
})
yield build_accordion(progress_steps)
# Save all to knowledge base (with deduplication)
merge_to_knowledge_base(prospects_found, contacts_found, emails_drafted)
# Build summary HTML
summary_html = f'''
β Discovery Complete!
Prospects Found
{len(prospects_found)}
Decision Makers
{len(contacts_found)}
Emails Drafted
{len(emails_drafted)}
'''
# Build detailed results section with collapsible prospect cards
results_html = ""
if prospects_found or contacts_found or emails_drafted:
results_html += """
π― Discovered Prospects
"""
for p in prospects_found:
p_name = p.get('name', 'Unknown')
p_name_lower = p_name.lower()
# Find contacts for this prospect - strict matching by exact company name
p_domain = p.get('domain', '').lower().replace('www.', '')
p_contacts = []
for c in contacts_found:
c_company = c.get("company", "").lower()
c_email = c.get("email", "").lower()
# Match by exact company name OR by email domain
if (c_company == p_name_lower or
p_name_lower == c_company or
(p_domain and p_domain in c_email)):
p_contacts.append(c)
# Find emails for this prospect - strict matching
p_emails = []
for e in emails_drafted:
e_company = e.get("prospect_company", "").lower()
e_to = e.get("to", "").lower()
if (e_company == p_name_lower or
p_name_lower == e_company or
(p_domain and p_domain in e_to)):
p_emails.append(e)
# Build contacts HTML
contacts_section = ""
if p_contacts:
contacts_section = "
βΉοΈ Note: No prospects were saved by the AI. Try running discovery again or adjusting your search criteria.
"""
# Yield final accordion with summary and results
yield build_accordion(progress_steps, is_loading=False, summary_html=summary_html + results_html)
progress(1.0)
return
elif event_type == "agent_max_iterations":
# Auto-generate emails if we have contacts but no emails
if contacts_found and not emails_drafted:
for c in contacts_found:
if c.get("email"):
contact_name = c.get("name", "").split()[0] if c.get("name") else "there"
full_name = c.get("name", "")
company = c.get("company", "your company")
title = c.get("title", "")
email_body = f"""Hi {contact_name},
I hope this message finds you well. I recently came across {company} and was genuinely impressed by the innovative work your team is doing.
As {title} at {company}, you're likely focused on driving growth and staying ahead of industry trends. That's exactly why I wanted to reach out.
At {client_name}, we specialize in helping companies like {company} achieve their strategic objectives. We've helped similar organizations:
β’ Streamline their operations and reduce costs
β’ Accelerate growth through innovative strategies
β’ Stay competitive in an evolving market
Would you be open to a brief 15-minute call this week to explore if there might be a fit?
Best regards,
{client_name} Team"""
emails_drafted.append({
"to": c.get("email"),
"subject": f"{contact_name}, quick question about {company}'s 2025 growth plans",
"body": email_body,
"prospect_company": company,
"contact_name": full_name,
"created_at": datetime.now().strftime("%Y-%m-%d %H:%M")
})
# Save what we found so far (with deduplication)
merge_to_knowledge_base(prospects_found, contacts_found, emails_drafted)
progress_steps.append({
"icon": "β±οΈ",
"icon_class": "warning",
"title": "Max iterations reached",
"detail": "Discovery stopped but results saved"
})
summary_html = f'''
β±οΈ Discovery Summary (Partial)
Prospects Found
{len(prospects_found)}
Decision Makers
{len(contacts_found)}
Emails Drafted
{len(emails_drafted)}
'''
yield build_accordion(progress_steps, is_loading=False, summary_html=summary_html)
return
elif event_type == "agent_error":
# Save what we found so far even on error (with deduplication)
merge_to_knowledge_base(prospects_found, contacts_found, emails_drafted)
error_msg = event.get("error", "Unknown error")
progress_steps.append({
"icon": "β",
"icon_class": "error",
"title": "Error occurred",
"detail": str(error_msg)[:100]
})
summary_html = f'''
β οΈ Discovery Interrupted
Prospects Found
{len(prospects_found)}
Decision Makers
{len(contacts_found)}
Emails Drafted
{len(emails_drafted)}
'''
yield build_accordion(progress_steps, is_loading=False, summary_html=summary_html)
return
except Exception as e:
logger.error(f"Discovery error: {e}")
# Save what we found (with deduplication)
merge_to_knowledge_base(prospects_found, contacts_found, emails_drafted)
progress_steps.append({
"icon": "β",
"icon_class": "error",
"title": "Discovery interrupted",
"detail": str(e)[:100]
})
summary_html = f'''
β οΈ Discovery Error
Saved {len(prospects_found)} prospects found so far.
'''
yield build_accordion(progress_steps, is_loading=False, summary_html=summary_html)
# ============================================================================
# AI CHAT - With MCP Tool Support
# ============================================================================
async def chat_with_ai_async(message: str, history: list, hf_token: str):
"""AI Chat powered by LLM with full MCP tool support"""
if not knowledge_base["client"]["name"]:
yield history + [[message, "β οΈ Please complete Setup first. Enter your company name in the Setup tab."]], ""
return
if not message.strip():
yield history, ""
return
token = get_hf_token(hf_token)
if not token:
yield history + [[message, "β οΈ Please enter your HuggingFace token in the Setup tab."]], ""
return
client_name = knowledge_base["client"]["name"]
client_info = knowledge_base["client"].get("raw_research", "")
# Always use LLM for all queries - this is a full AI assistant
try:
agent = AutonomousMCPAgentHF(
mcp_registry=mcp_registry,
hf_token=token,
provider=HF_PROVIDER,
model=HF_MODEL
)
# Build comprehensive context with all knowledge base data
prospects_detail = ""
if knowledge_base["prospects"]:
for i, p in enumerate(knowledge_base["prospects"][:10], 1):
p_name = p.get('name', 'Unknown')
p_name_lower = p_name.lower()
# Get contacts for this prospect
p_contacts = [c for c in knowledge_base["contacts"]
if p_name_lower in c.get("company", "").lower()
or c.get("company", "").lower() in p_name_lower]
contacts_str = ", ".join([f"{c.get('name')} ({c.get('email')})" for c in p_contacts]) if p_contacts else "No contacts"
prospects_detail += f"{i}. {p_name} - {p.get('industry', 'Unknown industry')}, Fit: {p.get('fit_score', 'N/A')}\n"
prospects_detail += f" Summary: {p.get('summary', 'No summary')[:100]}\n"
prospects_detail += f" Contacts: {contacts_str}\n"
else:
prospects_detail = "No prospects discovered yet."
emails_detail = ""
if knowledge_base["emails"]:
for e in knowledge_base["emails"][:5]:
emails_detail += f"- To: {e.get('to')} | Subject: {e.get('subject', 'No subject')[:50]}\n"
else:
emails_detail = "No emails drafted yet."
task = f"""You are an AI sales assistant for {client_name}. You are a helpful, knowledgeable assistant that can answer any question about the sales pipeline, prospects, contacts, and help with various sales tasks.
ABOUT {client_name}:
{client_info[:500] if client_info else "No company research available yet."}
CURRENT SALES PIPELINE:
======================
PROSPECTS ({len(knowledge_base['prospects'])}):
{prospects_detail}
CONTACTS ({len(knowledge_base['contacts'])}):
{len(knowledge_base['contacts'])} decision makers found across prospects.
DRAFTED EMAILS ({len(knowledge_base['emails'])}):
{emails_detail}
USER MESSAGE: {message}
INSTRUCTIONS:
- Answer the user's question helpfully and completely
- If they ask about prospects, contacts, or emails, use the data above
- If they ask you to search for something, use search_web tool
- If they ask you to draft an email, create a professional, personalized email
- If they ask for talking points, strategies, or recommendations, provide thoughtful, specific advice
- If they ask to find similar companies or new prospects, use search_web to research
- Be conversational and helpful - you're a knowledgeable sales assistant
- Don't say "I don't have that capability" - try to help with whatever they ask
- For follow-up questions, use context from the conversation
Respond naturally and helpfully to the user's message."""
response_text = ""
current_history = history + [[message, "π€ Thinking..."]]
yield current_history, ""
async for event in agent.run(task, max_iterations=12):
event_type = event.get("type")
if event_type == "tool_call":
tool = event.get("tool", "")
tool_input = event.get("input", {})
if tool == "search_web":
query = tool_input.get("query", "") if isinstance(tool_input, dict) else ""
response_text += f"π Searching: {query[:50]}...\n"
elif tool == "send_email":
response_text += f"βοΈ Drafting email...\n"
else:
response_text += f"π§ Using {tool}...\n"
current_history = history + [[message, response_text]]
yield current_history, ""
elif event_type == "tool_result":
tool = event.get("tool", "")
result = event.get("result", {})
# Capture data from tool results (with deduplication)
if tool == "save_prospect" and isinstance(result, dict):
prospect_data = {
"name": result.get("company_name", result.get("prospect_id", "Unknown")),
"domain": result.get("company_domain", result.get("domain", "")),
"fit_score": result.get("fit_score", 75),
"research_complete": True,
"discovered_at": datetime.now().strftime("%Y-%m-%d %H:%M")
}
merge_to_knowledge_base([prospect_data], [], [])
response_text += f"β Saved prospect: {prospect_data['name']}\n"
elif tool == "save_contact" and isinstance(result, dict):
merge_to_knowledge_base([], [result], [])
response_text += f"β Saved contact\n"
elif tool == "send_email" and isinstance(result, dict):
merge_to_knowledge_base([], [], [result])
response_text += f"β Email drafted\n"
elif tool == "search_web":
count = result.get("count", 0) if isinstance(result, dict) else 0
response_text += f"β Found {count} results\n"
current_history = history + [[message, response_text]]
yield current_history, ""
elif event_type == "thought":
thought = event.get("thought", "")
# Only show substantive thoughts, not processing messages
if thought and len(thought) > 50 and not thought.startswith("[Processing"):
# This is likely the AI's actual response
pass # We'll get this in agent_complete
elif event_type == "agent_complete":
final = event.get("final_answer", "")
if final and "CX AI Agent" not in final and "Powered by AI" not in final:
# Clean response - show just the final answer
if response_text:
response_text += "\n---\n\n"
response_text += final
elif not response_text:
response_text = "I've processed your request. Is there anything else you'd like to know?"
current_history = history + [[message, response_text]]
yield current_history, ""
return
elif event_type == "agent_error":
error = event.get("error", "Unknown error")
if "rate limit" in str(error).lower():
response_text += "\nβ οΈ Rate limit reached. Please wait a moment and try again."
else:
response_text += f"\nβ οΈ Error: {error}"
current_history = history + [[message, response_text]]
yield current_history, ""
return
elif event_type == "agent_max_iterations":
if not response_text:
response_text = "I'm still processing your request. The task may be complex - please try a simpler question or try again."
current_history = history + [[message, response_text]]
yield current_history, ""
return
# If we get here without returning
if not response_text:
response_text = "I processed your request. Let me know if you need anything else!"
yield history + [[message, response_text]], ""
except Exception as e:
logger.error(f"Chat agent error: {e}")
error_msg = str(e)
if "rate limit" in error_msg.lower() or "429" in error_msg:
yield history + [[message, "β οΈ Rate limit reached. Please wait a moment and try again."]], ""
else:
yield history + [[message, f"β οΈ Error: {error_msg}"]], ""
def chat_with_ai(message: str, history: list) -> tuple:
"""Chat function - handles queries using local data and templates"""
if not knowledge_base["client"]["name"]:
return history + [[message, "β οΈ Please complete Setup first. Enter your HuggingFace token and company name."]], ""
if not session_hf_token.get("token"):
return history + [[message, "β οΈ Please enter your HuggingFace token in the **Setup** tab first."]], ""
if not message.strip():
return history, ""
client_name = knowledge_base["client"]["name"]
msg_lower = message.lower()
def find_prospect_by_name(query: str):
"""Find prospect by exact or partial name match"""
query_lower = query.lower()
# First try exact match
for p in knowledge_base["prospects"]:
if p.get("name", "").lower() == query_lower:
return p
# Then try if prospect name contains query
for p in knowledge_base["prospects"]:
if query_lower in p.get("name", "").lower():
return p
# Then try if query contains prospect name
for p in knowledge_base["prospects"]:
p_name = p.get("name", "").lower()
if p_name in query_lower:
return p
# Finally try partial word match
query_words = set(query_lower.split())
for p in knowledge_base["prospects"]:
p_words = set(p.get("name", "").lower().split())
if query_words & p_words: # Any word in common
return p
return None
# Check for specific prospect mention using improved matching
mentioned_prospect = find_prospect_by_name(message)
# Handle "find decision makers" / "find contacts" for a known prospect
if any(kw in msg_lower for kw in ["find decision", "find contact", "who works at", "contacts at"]):
if mentioned_prospect:
p_name = mentioned_prospect["name"]
p_name_lower = p_name.lower()
contacts = [c for c in knowledge_base["contacts"]
if p_name_lower in c.get("company", "").lower()
or c.get("company", "").lower() in p_name_lower]
if contacts:
response = f"## π₯ Decision Makers at {p_name}\n\n"
for c in contacts:
response += f"**{c.get('name', 'Unknown')}** - {c.get('title', 'Unknown')}\n"
response += f" - Email: {c.get('email', 'Not available')}\n"
response += f" - Company: {c.get('company', p_name)}\n\n"
else:
response = f"No contacts found yet for **{p_name}**.\n\n"
response += "To find contacts, go to **Prospects Tab** and run **Find Prospects** again."
return history + [[message, response]], ""
# Handle "show email" - just viewing existing drafts
if any(kw in msg_lower for kw in ["show email", "existing email", "what email", "see email", "view email"]):
if mentioned_prospect:
p_name = mentioned_prospect["name"]
p_name_lower = p_name.lower()
existing_emails = [e for e in knowledge_base["emails"]
if p_name_lower in e.get("prospect_company", "").lower()]
if existing_emails:
email = existing_emails[0]
response = f"## βοΈ Existing Email Draft for {p_name}\n\n"
response += f"**To:** {email.get('to', 'N/A')}\n"
response += f"**Subject:** {email.get('subject', 'N/A')}\n\n"
response += f"---\n\n{email.get('body', 'No content')}\n\n"
response += "---\n\n*This email was drafted during prospect discovery.*"
else:
response = f"No existing email drafts found for **{p_name}**."
return history + [[message, response]], ""
# Handle "draft/write/compose email" - create custom email based on user's request
if any(kw in msg_lower for kw in ["draft", "write", "compose", "create email", "email to", "send email", "mail to"]):
if mentioned_prospect:
p_name = mentioned_prospect["name"]
p_name_lower = p_name.lower()
# Get contact info
contacts = [c for c in knowledge_base["contacts"]
if p_name_lower in c.get("company", "").lower()
or c.get("company", "").lower() in p_name_lower]
contact = contacts[0] if contacts else None
to_email = contact.get("email", f"contact@{p_name.lower().replace(' ', '')}.com") if contact else f"contact@{p_name.lower().replace(' ', '')}.com"
contact_name = contact.get("name", "").split()[0] if contact and contact.get("name") else "there"
contact_title = contact.get("title", "") if contact else ""
# Extract specific details from user's message
import re
# Check if this is a meeting request
is_meeting_request = any(kw in msg_lower for kw in ["meeting", "call", "demo", "schedule", "appointment"])
# Extract date/time info
date_match = re.search(r'(\d{1,2}(?:st|nd|rd|th)?\s+(?:jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*\s+\d{4}|\w+day(?:\s+next\s+week)?|\d{1,2}[/-]\d{1,2}[/-]\d{2,4})', msg_lower)
time_match = re.search(r'(\d{1,2}:\d{2}|\d{1,2}\s*(?:am|pm))', msg_lower)
duration_match = re.search(r'(\d+)\s*(?:min|minute|hour)', msg_lower)
date_str = date_match.group(1).title() if date_match else ""
time_str = time_match.group(1) if time_match else ""
duration_str = duration_match.group(0) if duration_match else ""
# Extract the purpose/topic from the message
# Remove common words to find the custom content
custom_content = message
for word in ["draft", "write", "compose", "email", "mail", "to", p_name.lower(), "asking", "that", "can", "we", "a", "an", "the", "for", "about"]:
custom_content = re.sub(rf'\b{word}\b', '', custom_content, flags=re.IGNORECASE)
custom_content = ' '.join(custom_content.split()).strip()
# Generate custom email based on context
response = f"## βοΈ Custom Email Draft for {p_name}\n\n"
response += f"**To:** {to_email}\n"
if is_meeting_request:
# Meeting request email
subject = f"Meeting Request: {client_name} x {p_name}"
if date_str:
subject = f"Meeting Request for {date_str} - {client_name} x {p_name}"
response += f"**Subject:** {subject}\n\n"
response += f"---\n\n"
response += f"Dear {contact_name},\n\n"
response += f"I hope this email finds you well.\n\n"
response += f"I'm reaching out from {client_name} regarding a potential collaboration with {p_name}. "
response += f"Based on our research, we believe there's a strong synergy between our companies, "
response += f"particularly in the {mentioned_prospect.get('industry', 'your industry')} space.\n\n"
if date_str or time_str or duration_str:
response += f"I would like to propose a meeting"
if date_str:
response += f" on **{date_str}**"
if time_str:
response += f" at **{time_str}**"
if duration_str:
response += f" for **{duration_str}**"
response += f" to discuss how {client_name} can help {p_name} achieve its goals.\n\n"
else:
response += f"Would you be available for a brief call this week to discuss how {client_name} can support {p_name}'s growth?\n\n"
response += f"During our conversation, I'd love to explore:\n"
response += f"- How {client_name}'s solutions align with {p_name}'s current initiatives\n"
response += f"- Specific ways we can add value to your {mentioned_prospect.get('industry', 'business')}\n"
response += f"- Next steps for a potential partnership\n\n"
response += f"Please let me know if this time works for you, or suggest an alternative that fits your schedule.\n\n"
else:
# General outreach with custom content
subject = f"{client_name} + {p_name}: Let's Connect"
response += f"**Subject:** {subject}\n\n"
response += f"---\n\n"
response += f"Dear {contact_name},\n\n"
response += f"I'm reaching out from {client_name} regarding {p_name}.\n\n"
if custom_content:
response += f"{custom_content}\n\n"
response += f"Based on our research into {p_name}'s work in {mentioned_prospect.get('industry', 'your industry')}, "
response += f"we believe {client_name} can provide significant value.\n\n"
response += f"**About {p_name}:** {mentioned_prospect.get('summary', '')}\n\n"
response += f"**Why we're reaching out:** {mentioned_prospect.get('fit_reason', 'We see great potential for collaboration.')}\n\n"
response += f"Would you be open to a conversation about how we can work together?\n\n"
response += f"Best regards,\n"
response += f"[Your Name]\n"
response += f"{client_name}\n\n"
response += f"---\n\n"
response += f"*π This is a custom draft based on your request. Edit as needed before sending.*"
return history + [[message, response]], ""
# Handle "suggest talking points" for a prospect
if any(kw in msg_lower for kw in ["talking point", "suggest", "recommend", "strategy"]):
if mentioned_prospect:
p_name = mentioned_prospect["name"]
response = f"## π‘ Talking Points for {p_name}\n\n"
response += f"**About {p_name}:**\n"
response += f"- Industry: {mentioned_prospect.get('industry', 'Unknown')}\n"
response += f"- {mentioned_prospect.get('summary', 'No summary available')}\n\n"
response += f"**Why they're a fit for {client_name}:**\n"
response += f"- {mentioned_prospect.get('fit_reason', 'Matches target customer profile')}\n\n"
response += f"**Suggested talking points:**\n"
response += f"1. Reference their focus on {mentioned_prospect.get('industry', 'their industry')}\n"
response += f"2. Highlight how {client_name} can help with scalability\n"
response += f"3. Mention success stories from similar companies\n"
response += f"4. Propose a specific next step (demo, call, pilot)\n"
return history + [[message, response]], ""
# Handle "research [prospect]" or "analyze [prospect]" - show detailed info
if any(kw in msg_lower for kw in ["research", "analyze", "details about", "info on", "information about"]):
if mentioned_prospect:
p_name = mentioned_prospect["name"]
p_name_lower = p_name.lower()
# Get contacts and emails for this prospect
contacts = [c for c in knowledge_base["contacts"]
if p_name_lower in c.get("company", "").lower()
or c.get("company", "").lower() in p_name_lower]
emails = [e for e in knowledge_base["emails"]
if p_name_lower in e.get("prospect_company", "").lower()]
response = f"## π Research: {p_name}\n\n"
response += f"### Company Overview\n"
response += f"- **Industry:** {mentioned_prospect.get('industry', 'Unknown')}\n"
response += f"- **Fit Score:** {mentioned_prospect.get('fit_score', 'N/A')}/100\n"
response += f"- **Summary:** {mentioned_prospect.get('summary', 'No summary available')}\n\n"
response += f"### Why They're a Good Fit for {client_name}\n"
response += f"{mentioned_prospect.get('fit_reason', 'Matches target customer profile')}\n\n"
response += f"### Decision Makers ({len(contacts)})\n"
if contacts:
for c in contacts:
response += f"- **{c.get('name', 'Unknown')}** - {c.get('title', 'Unknown')}\n"
response += f" - Email: {c.get('email', 'N/A')}\n"
else:
response += "No contacts found yet.\n"
response += f"\n### Outreach Status\n"
if emails:
response += f"β {len(emails)} email(s) drafted\n"
for e in emails:
response += f"- To: {e.get('to', 'N/A')} - \"{e.get('subject', 'No subject')[:40]}...\"\n"
else:
response += "β³ No emails drafted yet\n"
return history + [[message, response]], ""
# Handle "find competitors" or "competitors to"
if any(kw in msg_lower for kw in ["competitor", "similar to", "like "]):
if mentioned_prospect:
p_name = mentioned_prospect["name"]
industry = mentioned_prospect.get('industry', 'Unknown')
response = f"## π’ Finding Similar Companies to {p_name}\n\n"
response += f"**{p_name}** is in the **{industry}** industry.\n\n"
response += f"To find more companies similar to {p_name}:\n\n"
response += f"1. Go to **Prospects Tab**\n"
response += f"2. The AI will search for companies in {industry}\n"
response += f"3. It will identify competitors and similar businesses\n\n"
response += f"**Currently in your pipeline:**\n"
other_in_industry = [p for p in knowledge_base["prospects"]
if p.get("industry", "").lower() == industry.lower() and p.get("name") != p_name]
if other_in_industry:
response += f"Other {industry} prospects:\n"
for p in other_in_industry:
response += f"- {p.get('name')} (Fit: {p.get('fit_score', 'N/A')})\n"
else:
response += f"No other {industry} prospects found yet.\n"
return history + [[message, response]], ""
# For generic "search for new" or "discover new" - guide to prospects tab
if any(kw in msg_lower for kw in ["search for new", "find new", "discover new", "look for new"]):
response = f"""π **Search for New Prospects**
To discover new companies, use the **Prospects Tab**:
1. Go to **Prospects** tab
2. Enter the number of prospects to find
3. Click **"Find Prospects & Contacts"**
The AI will:
- Search for companies matching {client_name}'s target market
- Find decision makers at each company
- Draft personalized outreach emails
**Currently in your pipeline:**
- Prospects: {len(knowledge_base['prospects'])}
- Contacts: {len(knowledge_base['contacts'])}
- Emails: {len(knowledge_base['emails'])}
"""
return history + [[message, response]], ""
# For simple queries, use local knowledge base lookup
response = get_local_response(message, client_name)
return history + [[message, response]], ""
def get_local_response(message: str, client_name: str) -> str:
"""Handle simple queries locally without AI agent"""
msg_lower = message.lower()
# Detect user intent and respond accordingly
response = ""
# Intent: List prospects
if any(kw in msg_lower for kw in ["list prospect", "show prospect", "all prospect", "prospects"]):
if knowledge_base["prospects"]:
response = f"## π― Prospects for {client_name}\n\n"
for i, p in enumerate(knowledge_base["prospects"], 1):
response += f"**{i}. {p.get('name', 'Unknown')}**\n"
response += f" - Industry: {p.get('industry', 'Unknown')}\n"
response += f" - Fit Score: {p.get('fit_score', 'N/A')}/100\n"
if p.get('summary'):
response += f" - Summary: {p.get('summary', '')[:150]}...\n" if len(p.get('summary', '')) > 150 else f" - Summary: {p.get('summary', '')}\n"
response += "\n"
else:
response = "No prospects discovered yet. Go to the **Discovery** tab and click **Find Prospects & Contacts** to discover potential customers."
# Intent: List contacts / decision makers
elif any(kw in msg_lower for kw in ["contact", "decision maker", "who", "email address", "reach"]):
# Check if asking about specific prospect
specific_prospect = None
for p in knowledge_base["prospects"]:
if p.get("name", "").lower() in msg_lower:
specific_prospect = p
break
if specific_prospect:
prospect_contacts = [c for c in knowledge_base["contacts"] if c.get("company", "").lower() == specific_prospect["name"].lower()]
if prospect_contacts:
response = f"## π₯ Decision Makers at {specific_prospect['name']}\n\n"
for c in prospect_contacts:
response += f"**{c.get('name', 'Unknown')}**\n"
response += f" - Title: {c.get('title', 'Unknown')}\n"
response += f" - Email: {c.get('email', 'Not available')}\n"
if c.get('linkedin'):
response += f" - LinkedIn: {c.get('linkedin')}\n"
response += "\n"
else:
response = f"No contacts found for **{specific_prospect['name']}** yet."
elif knowledge_base["contacts"]:
response = f"## π₯ All Decision Makers\n\n"
for c in knowledge_base["contacts"]:
response += f"**{c.get('name', 'Unknown')}** - {c.get('title', 'Unknown')}\n"
response += f" - Company: {c.get('company', 'Unknown')}\n"
response += f" - Email: {c.get('email', 'Not available')}\n\n"
else:
response = "No contacts discovered yet. Run **Find Prospects** to discover decision makers."
# Intent: Show emails
elif any(kw in msg_lower for kw in ["email", "draft", "outreach", "message"]):
specific_prospect = None
for p in knowledge_base["prospects"]:
if p.get("name", "").lower() in msg_lower:
specific_prospect = p
break
if specific_prospect:
prospect_emails = [e for e in knowledge_base["emails"] if specific_prospect["name"].lower() in e.get("prospect_company", "").lower()]
if prospect_emails:
response = f"## βοΈ Emails for {specific_prospect['name']}\n\n"
for e in prospect_emails:
response += f"**To:** {e.get('to', 'Unknown')}\n"
response += f"**Subject:** {e.get('subject', 'No subject')}\n\n"
response += f"```\n{e.get('body', 'No content')}\n```\n\n"
else:
response = f"No emails drafted for **{specific_prospect['name']}** yet."
elif knowledge_base["emails"]:
response = "## βοΈ All Drafted Emails\n\n"
for e in knowledge_base["emails"]:
response += f"**To:** {e.get('to', 'Unknown')} ({e.get('prospect_company', 'Unknown')})\n"
response += f"**Subject:** {e.get('subject', 'No subject')}\n\n"
else:
response = "No emails drafted yet. Run **Find Prospects** to have AI draft outreach emails."
# Intent: Tell me about / describe prospect
elif any(kw in msg_lower for kw in ["tell me about", "describe", "info about", "details", "about"]):
specific_prospect = None
for p in knowledge_base["prospects"]:
if p.get("name", "").lower() in msg_lower:
specific_prospect = p
break
if specific_prospect:
response = f"## π’ {specific_prospect['name']}\n\n"
response += f"**Industry:** {specific_prospect.get('industry', 'Unknown')}\n"
response += f"**Fit Score:** {specific_prospect.get('fit_score', 'N/A')}/100\n\n"
if specific_prospect.get('summary'):
response += f"**Summary:**\n{specific_prospect.get('summary')}\n\n"
if specific_prospect.get('fit_reason'):
response += f"**Why they're a good fit:**\n{specific_prospect.get('fit_reason')}\n\n"
# Show contacts for this prospect
prospect_contacts = [c for c in knowledge_base["contacts"] if c.get("company", "").lower() == specific_prospect["name"].lower()]
if prospect_contacts:
response += f"**Decision Makers ({len(prospect_contacts)}):**\n"
for c in prospect_contacts:
response += f"- {c.get('name', 'Unknown')} - {c.get('title', '')} ({c.get('email', 'no email')})\n"
elif knowledge_base["prospects"]:
response = "Which prospect would you like to know about?\n\n**Available prospects:**\n"
for p in knowledge_base["prospects"]:
response += f"- {p.get('name', 'Unknown')}\n"
else:
response = "No prospects discovered yet. Run **Find Prospects** first."
# Intent: Summary / overview
elif any(kw in msg_lower for kw in ["summary", "overview", "status", "pipeline", "how many"]):
response = f"## π {client_name} Sales Pipeline Summary\n\n"
response += f"| Metric | Count |\n"
response += f"|--------|-------|\n"
response += f"| Prospects | {len(knowledge_base['prospects'])} |\n"
response += f"| Decision Makers | {len(knowledge_base['contacts'])} |\n"
response += f"| Emails Drafted | {len(knowledge_base['emails'])} |\n\n"
if knowledge_base["prospects"]:
response += "**Prospects:**\n"
for p in knowledge_base["prospects"]:
response += f"- {p.get('name', 'Unknown')} (Fit: {p.get('fit_score', 'N/A')})\n"
# Intent: Help / what can you do
elif any(kw in msg_lower for kw in ["help", "what can", "how do", "?"]):
response = f"""## π¬ {client_name} Sales Assistant
I can help you with information about your sales pipeline. Try asking:
**About Prospects:**
- "List all prospects"
- "Tell me about [prospect name]"
- "Show prospect details"
**About Contacts:**
- "Who are the decision makers?"
- "Show contacts for [prospect name]"
- "List all contacts"
**About Emails:**
- "Show drafted emails"
- "What emails do we have for [prospect name]?"
**Pipeline Overview:**
- "Give me a summary"
- "How many prospects do we have?"
- "Pipeline status"
"""
# Default: Try to be helpful
else:
prospects_list = ", ".join([p.get("name", "Unknown") for p in knowledge_base["prospects"]]) if knowledge_base["prospects"] else "None yet"
response = f"""I'm not sure what you're asking. Here's what I know:
**Current Pipeline:**
- Prospects: {len(knowledge_base["prospects"])} ({prospects_list})
- Contacts: {len(knowledge_base["contacts"])}
- Emails: {len(knowledge_base["emails"])}
Try asking:
- "List prospects"
- "Tell me about [prospect name]"
- "Show contacts"
- "Show emails"
- "Give me a summary"
"""
return response
# ============================================================================
# HANDOFF PACKET
# ============================================================================
def generate_handoff_packet(prospect_name: str) -> str:
if not prospect_name:
return "β οΈ Please select a prospect."
prospect = next((p for p in knowledge_base["prospects"] if p["name"] == prospect_name), None)
if not prospect:
return f"β οΈ Prospect '{prospect_name}' not found."
# Case-insensitive contact matching with partial match support
prospect_name_lower = prospect_name.lower()
contacts = [c for c in knowledge_base["contacts"]
if prospect_name_lower in c.get("company", "").lower()
or c.get("company", "").lower() in prospect_name_lower]
# Also match emails for this prospect (case-insensitive, partial match)
emails_for_prospect = [e for e in knowledge_base["emails"]
if prospect_name_lower in e.get("prospect_company", "").lower()
or e.get("prospect_company", "").lower() in prospect_name_lower]
email = emails_for_prospect[0] if emails_for_prospect else None
# If no contacts found but we have an email, extract contact from email
if not contacts and email:
email_to = email.get("to", "")
if email_to:
# Try to extract name from email body or use email
email_body = email.get("body", "")
# Look for "Dear [Name]" pattern
import re
name_match = re.search(r'Dear\s+([A-Z][a-z]+)', email_body)
contact_name = name_match.group(1) if name_match else email_to.split('@')[0].title()
contacts = [{
"name": contact_name,
"email": email_to,
"title": "Contact",
"company": prospect_name
}]
client_name = knowledge_base["client"]["name"]
packet = f"""# π Sales Handoff Packet
## {prospect["name"]}
**Prepared for:** {client_name}
**Date:** {datetime.now().strftime("%Y-%m-%d")}
---
## 1. Company Overview
{prospect.get("summary", "No summary available.")}
**Industry:** {prospect.get("industry", "Unknown")}
**Fit Score:** {prospect.get("fit_score", "N/A")}/100
---
## 2. Why They're a Good Fit
{prospect.get("fit_reason", "Matches ideal customer profile.")}
---
## 3. Decision Makers ({len(contacts)})
"""
for c in contacts:
packet += f"- **{c.get('name', 'Unknown')}** - {c.get('title', 'Contact')}"
if c.get('email'):
packet += f" ({c.get('email')})"
packet += "\n"
if not contacts:
packet += "No contacts identified yet.\n"
packet += f"""
---
## 4. Recommended Approach
1. Lead with {client_name}'s value proposition
2. Reference their specific challenges
3. Propose concrete next step (demo, call)
---
## 5. Drafted Email
"""
if email:
packet += f"""**To:** {email.get("to", "N/A")}
**Subject:** {email.get("subject", "N/A")}
---
{email.get("body", "No email body.")}
"""
else:
packet += "No email drafted yet.\n"
packet += f"""
---
*Generated by CX AI Agent for {client_name}*
"""
return packet
def get_prospect_choices():
return [p["name"] for p in knowledge_base["prospects"]] if knowledge_base["prospects"] else []
# ============================================================================
# GRADIO UI
# ============================================================================
def get_logo_base64():
"""Load logo image as base64 for embedding in HTML"""
logo_path = Path(__file__).parent / "assets" / "cx_ai_agent_logo_512.png"
if logo_path.exists():
with open(logo_path, "rb") as f:
return base64.b64encode(f.read()).decode("utf-8")
return None
def get_favicon_base64():
"""Load favicon as base64 for embedding"""
favicon_path = Path(__file__).parent / "assets" / "cx_ai_agent_favicon_32.png"
if favicon_path.exists():
with open(favicon_path, "rb") as f:
return base64.b64encode(f.read()).decode("utf-8")
return None
def create_app():
# Load logo as base64
logo_b64 = get_logo_base64()
favicon_b64 = get_favicon_base64()
# Build sidebar logo HTML
sidebar_logo = f'' if logo_b64 else '
CX
'
# Custom head HTML
favicon_html = f'' if favicon_b64 else ''
head_html = f"""
{favicon_html}
"""
with gr.Blocks(
title="CX AI Agent - B2B Sales Intelligence",
theme=gr.themes.Soft(primary_hue="blue", secondary_hue="slate", neutral_hue="slate"),
css=ENTERPRISE_CSS,
head=head_html
) as demo:
# ===== SIDEBAR (HTML) =====
gr.HTML(f"""
CX AI Agent
{sidebar_logo}
CX AI Agent
""")
# ===== MAIN CONTENT WRAPPER =====
with gr.Column(elem_classes="main-wrapper"):
# Hidden page selector for navigation state
page_selector = gr.Textbox(value="setup", visible=False, elem_id="page-selector")
# Navigation buttons row (hidden on desktop, visible on mobile as fallback)
with gr.Row(elem_classes="nav-buttons-row", visible=True):
btn_setup = gr.Button("βοΈ Setup", elem_id="btn-setup", size="sm")
btn_dashboard = gr.Button("π Dashboard", elem_id="btn-dashboard", size="sm")
btn_discovery = gr.Button("π Discovery", elem_id="btn-discovery", size="sm")
btn_prospects = gr.Button("π― Prospects", elem_id="btn-prospects", size="sm")
btn_contacts = gr.Button("π₯ Contacts", elem_id="btn-contacts", size="sm")
btn_emails = gr.Button("βοΈ Emails", elem_id="btn-emails", size="sm")
btn_chat = gr.Button("π¬ Chat", elem_id="btn-chat", size="sm")
btn_about = gr.Button("βΉοΈ About", elem_id="btn-about", size="sm")
# ===== SETUP PAGE =====
with gr.Column(visible=True, elem_id="page-setup") as setup_page:
gr.HTML("""
βοΈ Setup
Configure your company and API credentials
π
Getting Started
Complete these steps to start finding prospects:
HuggingFace Token - Required for AI-powered research and email drafting
Serper API Key - Optional, enables real-time web search for company info
Company Name - Your company name helps AI find relevant prospects
""")
with gr.Row():
with gr.Column(scale=1):
gr.HTML("""
""")
hf_token_input = gr.Textbox(
label="HuggingFace Token",
placeholder="hf_xxxxxxxxxx",
type="password"
)
serper_key_input = gr.Textbox(
label="Serper API Key (Optional)",
placeholder="For web search - get at serper.dev",
type="password"
)
gr.HTML("""
π’ Your Company
AI will research your company and find matching prospects.
""")
client_name_input = gr.Textbox(label="Company Name", placeholder="e.g., Acme Corp")
with gr.Row():
setup_btn = gr.Button("π Setup Company", variant="primary", size="lg")
reset_btn = gr.Button("ποΈ Reset", variant="stop", size="sm")
with gr.Column(scale=2):
setup_output = gr.Markdown("*Enter your credentials and company name to begin.*")
# ===== DASHBOARD PAGE =====
with gr.Column(visible=True, elem_id="page-dashboard", elem_classes="page-hidden") as dashboard_page:
gr.HTML("""
π Dashboard
Overview of your sales pipeline
π
Pipeline Overview
Track your progress at a glance. The dashboard shows real-time counts of prospects discovered, contacts found, and emails drafted. Click "Refresh" to update the stats after running Discovery.
")
# ===== PROSPECTS PAGE =====
with gr.Column(visible=True, elem_id="page-prospects", elem_classes="page-hidden") as prospects_page:
gr.HTML("""
π― Prospects
Companies discovered by AI
π’
Your Prospect Companies
This list shows all companies found by the AI. Each prospect includes company details, industry, and a fit score (0-100) indicating how well they match your ideal customer profile. Higher scores = better fit!
AI finds key decision-makers (CEOs, VPs, Founders, Directors) at each prospect company. Contact info includes name, title, email, and company. Only verified contacts with real email addresses are shown.
Each email is personalized based on the prospect's company, industry, and any pain points discovered during research. Review and customize before sending. Emails are designed to start conversations, not close deals.
""")
refresh_emails_btn = gr.Button("π Refresh", variant="secondary", size="sm")
emails_list = gr.HTML(get_emails_html())
# ===== AI CHAT PAGE =====
with gr.Column(visible=True, elem_id="page-chat", elem_classes="page-hidden") as chat_page:
gr.HTML("""
π¬ AI Chat
AI-powered communication hub
""")
with gr.Tabs(elem_classes="chat-subtabs"):
# ----- SUB-TAB 1: Internal Sales Assistant -----
with gr.Tab("π― Sales Assistant", elem_id="tab-sales-assistant"):
gr.HTML("""
π€
Your AI Sales Assistant
Chat with AI to research companies, draft emails, get talking points, or manage your pipeline. The AI has access to all your prospect data and can perform web searches for real-time info.
"Search for DTC fashion brands that raised Series A"
"Draft an email to the CEO of Warby Parker"
"Give me talking points for my call with Glossier"
"Summary of all prospects and their status"
""")
# ----- SUB-TAB 2: Prospect-Facing AI Chat -----
with gr.Tab("π€ Prospect Chat Demo", elem_id="tab-prospect-chat"):
gr.HTML("""
π¬
Prospect Communication Demo
This demonstrates how prospects can interact with your company's AI assistant. The AI can answer questions about your products/services, qualify leads, schedule meetings, and escalate to human agents when needed.
""")
prospect_chatbot = gr.Chatbot(
value=[],
height=350,
label="Prospect Chat",
avatar_images=(None, "https://api.dicebear.com/7.x/bottts/svg?seed=cx-agent")
)
with gr.Row():
prospect_input = gr.Textbox(
label="Prospect Message",
placeholder="Hi, I'm interested in learning more about your services...",
lines=1,
scale=4
)
prospect_send_btn = gr.Button("Send", variant="primary", scale=1)
with gr.Row():
with gr.Column(scale=2):
gr.HTML("""
π Demo Scenario
You are a prospect visiting the client's website. The AI will:
Answer questions about products and services
Qualify you as a lead based on your needs
Offer to schedule a meeting with sales
Escalate complex inquiries to human agents
""")
with gr.Column(scale=1):
gr.HTML("""
β‘ Quick Actions
""")
generate_handoff_btn = gr.Button("π Generate Handoff Packet", variant="secondary", size="sm")
escalate_btn = gr.Button("π¨ Escalate to Human", variant="stop", size="sm")
schedule_btn = gr.Button("π Schedule Meeting", variant="secondary", size="sm")
handoff_output = gr.Markdown(visible=False, elem_classes="handoff-packet")
# ===== ABOUT US PAGE =====
with gr.Column(visible=True, elem_id="page-about", elem_classes="page-hidden") as about_page:
gr.HTML("""
βΉοΈ About Us
Learn more about CX AI Agent
""")
gr.Markdown("""
# π€ CX AI Agent - B2B Sales Intelligence Platform
[](https://github.com)
[](https://huggingface.co)
[](https://gradio.app)
> **π MCP in Action Track - Enterprise Applications**
>
> Tag: `mcp-in-action-track-enterprise`
---
## π Overview
**CX AI Agent** is an AI-powered B2B sales automation platform that helps sales teams discover prospects, find decision-makers, and draft personalized outreach emailsβall powered by autonomous AI agents using the Model Context Protocol (MCP).
### π― Key Features
| Feature | Description |
|---------|-------------|
| **π AI Discovery** | Automatically find and research prospect companies matching your ideal customer profile |
| **π₯ Contact Finder** | Locate decision-makers (CEOs, VPs, Founders) with verified email addresses |
| **βοΈ Email Drafting** | Generate personalized cold outreach emails based on company research |
| **π¬ AI Chat** | Interactive assistant for pipeline management and real-time research |
| **π€ Prospect Chat** | Demo of prospect-facing AI with handoff & escalation capabilities |
| **π Dashboard** | Real-time pipeline metrics and progress tracking |
---
## ποΈ Architecture
```
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β CX AI Agent β
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ€
β βββββββββββββββ βββββββββββββββ βββββββββββββββ β
β β Gradio β β Autonomousβ β MCP β β
β β UI ββββ Agent ββββ Servers β β
β βββββββββββββββ βββββββββββββββ βββββββββββββββ β
β β β β β
β βΌ βΌ βΌ β
β βββββββββββββββββββββββββββββββββββββββββββββββββββ β
β β MCP Tool Definitions β β
β β β’ Search (Web, News) β β
β β β’ Store (Prospects, Contacts, Facts) β β
β β β’ Email (Send, Thread Management) β β
β β β’ Calendar (Meeting Slots, Invites) β β
β βββββββββββββββββββββββββββββββββββββββββββββββββββ β
ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
```
---
## π Getting Started
### Prerequisites
- Python 3.8+
- HuggingFace API Token ([Get one free](https://huggingface.co/settings/tokens))
- Serper API Key (Optional, for web search)
### Quick Start
1. **Setup**: Enter your API credentials and company name
2. **Discover**: Let AI find prospects matching your profile
3. **Review**: Check discovered companies and contacts
4. **Engage**: Use AI-drafted emails for outreach
---
## π§ MCP Tools Available
### Search MCP Server
- `search_web` - Search the web for company information
- `search_news` - Find recent news about companies
### Store MCP Server
- `save_prospect` / `get_prospect` / `list_prospects` - Manage prospects
- `save_company` / `get_company` - Store company data
- `save_contact` / `list_contacts_by_domain` - Manage contacts
- `save_fact` - Store research insights
- `discover_prospects_with_contacts` - Full discovery pipeline
- `find_verified_contacts` - Find decision-makers
- `check_suppression` - Compliance checking
### Email MCP Server
- `send_email` - Send outreach emails
- `get_email_thread` - Retrieve conversation history
### Calendar MCP Server
- `suggest_meeting_slots` - Generate available times
- `generate_calendar_invite` - Create .ics files
---
## π Prospect Chat Demo
The **Prospect Chat Demo** tab showcases how prospects can interact with your company's AI:
- **Lead Qualification**: AI asks qualifying questions to understand prospect needs
- **Handoff Packets**: Generate comprehensive summaries for human sales reps
- **Escalation Flows**: Automatically escalate complex inquiries to humans
- **Meeting Scheduling**: Integrate with calendar for instant booking
---
## π Technology Stack
| Component | Technology |
|-----------|------------|
| **Frontend** | Gradio 5.x |
| **AI Model** | Qwen3-32B via HuggingFace |
| **Protocol** | Model Context Protocol (MCP) |
| **Search** | Serper API |
| **Language** | Python 3.8+ |
---
## π License
This project is open source and available under the MIT License.
---
## π Acknowledgments
- **Anthropic** - Model Context Protocol specification
- **HuggingFace** - AI model hosting and inference
- **Gradio** - UI framework
- **Serper** - Web search API
---
## π¨βπ» Developer
**Syed Muzakkir Hussain**
[](https://huggingface.co/muzakkirhussain011)
[https://huggingface.co/muzakkirhussain011](https://huggingface.co/muzakkirhussain011)
---
**Built with β€οΈ by [Syed Muzakkir Hussain](https://huggingface.co/muzakkirhussain011) for the Gradio Agents & MCP Hackathon 2025**
`mcp-in-action-track-enterprise`
""")
# Footer
gr.HTML("""
""")
# ===== NAVIGATION HANDLERS =====
all_pages = [setup_page, dashboard_page, discovery_page, prospects_page, contacts_page, emails_page, chat_page, about_page]
def show_page(page_name):
"""Return visibility updates for all pages"""
pages = {
"setup": [True, False, False, False, False, False, False, False],
"dashboard": [False, True, False, False, False, False, False, False],
"discovery": [False, False, True, False, False, False, False, False],
"prospects": [False, False, False, True, False, False, False, False],
"contacts": [False, False, False, False, True, False, False, False],
"emails": [False, False, False, False, False, True, False, False],
"chat": [False, False, False, False, False, False, True, False],
"about": [False, False, False, False, False, False, False, True],
}
visibility = pages.get(page_name, pages["setup"])
return [gr.update(visible=v) for v in visibility]
# When page_selector textbox changes, update page visibility
page_selector.change(fn=show_page, inputs=[page_selector], outputs=all_pages)
# Connect navigation buttons to pages
btn_setup.click(fn=lambda: show_page("setup"), outputs=all_pages)
btn_dashboard.click(fn=lambda: show_page("dashboard"), outputs=all_pages)
btn_discovery.click(fn=lambda: show_page("discovery"), outputs=all_pages)
btn_prospects.click(fn=lambda: show_page("prospects"), outputs=all_pages)
btn_contacts.click(fn=lambda: show_page("contacts"), outputs=all_pages)
btn_emails.click(fn=lambda: show_page("emails"), outputs=all_pages)
btn_chat.click(fn=lambda: show_page("chat"), outputs=all_pages)
btn_about.click(fn=lambda: show_page("about"), outputs=all_pages)
# Navigation JavaScript is now in head_html for earlier loading
# ===== EVENT HANDLERS =====
# Setup button - run setup and then update status indicators
setup_btn.click(
fn=setup_client_company,
inputs=[client_name_input, hf_token_input, serper_key_input],
outputs=[setup_output]
).then(
fn=lambda: (get_client_status_html(), get_client_status_html()),
outputs=[client_status, client_status_2]
)
reset_btn.click(
fn=reset_all_data,
outputs=[prospects_stat, contacts_stat, emails_stat, client_status, prospects_list, emails_list,
contacts_list, client_name_input, setup_output, discovery_output]
)
def refresh_dashboard():
stats = get_dashboard_stats()
return stats[0], stats[1], stats[2], stats[3]
refresh_btn.click(fn=refresh_dashboard, outputs=[prospects_stat, contacts_stat, emails_stat, client_status])
# Discover prospects and then update all lists
discover_btn.click(
fn=discover_prospects,
inputs=[num_prospects],
outputs=[discovery_output]
).then(
fn=lambda: (get_prospects_html(), get_contacts_html(), get_emails_html()),
outputs=[prospects_list, contacts_list, emails_list]
).then(
fn=refresh_dashboard,
outputs=[prospects_stat, contacts_stat, emails_stat, client_status]
)
refresh_prospects_btn.click(fn=get_prospects_html, outputs=[prospects_list])
refresh_contacts_btn.click(fn=get_contacts_html, outputs=[contacts_list])
refresh_emails_btn.click(fn=get_emails_html, outputs=[emails_list])
# Async chat wrapper that uses session token
async def chat_async_wrapper(message, history):
token = session_hf_token.get("token", "")
final_result = (history, "")
async for result in chat_with_ai_async(message, history, token):
final_result = result
return final_result
send_btn.click(fn=chat_async_wrapper, inputs=[chat_input, chatbot], outputs=[chatbot, chat_input])
chat_input.submit(fn=chat_async_wrapper, inputs=[chat_input, chatbot], outputs=[chatbot, chat_input])
# ===== PROSPECT CHAT HANDLERS =====
async def prospect_chat_wrapper(message, history):
"""Handle prospect-facing chat with company representative AI"""
if not message.strip():
return history, ""
# Get client company info for context
client_info = knowledge_base["client"].get("name") or "Our Company"
# Build prospect-facing system context
system_context = f"""You are an AI assistant representing {client_info}. You are speaking with a potential prospect who is interested in learning about the company's products and services.
Your role is to:
1. Answer questions about the company professionally and helpfully
2. Qualify the prospect by understanding their needs, company size, and timeline
3. Offer to schedule meetings with sales representatives when appropriate
4. Escalate complex technical or pricing questions to human agents
Be friendly, professional, and helpful. Focus on understanding the prospect's needs."""
history = history + [[message, None]]
# Use the AI to generate response
token = session_hf_token.get("token", "")
if token:
try:
from huggingface_hub import InferenceClient
client = InferenceClient(token=token)
messages = [{"role": "system", "content": system_context}]
for h in history[:-1]:
if h[0]:
messages.append({"role": "user", "content": h[0]})
if h[1]:
messages.append({"role": "assistant", "content": h[1]})
messages.append({"role": "user", "content": message})
response = client.chat_completion(
model="Qwen/Qwen2.5-72B-Instruct",
messages=messages,
max_tokens=500
)
reply = response.choices[0].message.content
except Exception as e:
reply = f"I apologize, I'm having trouble connecting right now. Please try again or contact us directly. (Error: {str(e)[:50]})"
else:
reply = f"Thank you for your interest in {client_info}! I'd be happy to help you learn more about our solutions. What specific challenges are you looking to address?"
history[-1][1] = reply
return history, ""
def generate_handoff_packet(chat_history):
"""Generate a handoff packet from the prospect conversation"""
if not chat_history:
return gr.update(visible=True, value="**β οΈ No conversation to generate handoff from.** Start a conversation first.")
# Extract key info from conversation
conversation_text = "\n".join([f"Prospect: {h[0]}\nAgent: {h[1]}" for h in chat_history if h[0] and h[1]])
client_name = knowledge_base["client"].get("name") or "Unknown Client"
packet = f"""
## π Handoff Packet
**Generated:** {datetime.now().strftime("%Y-%m-%d %H:%M")}
**Client Company:** {client_name}
---
### π Conversation Summary
{len(chat_history)} messages exchanged with prospect.
### π¬ Full Conversation Log
```
{conversation_text[:1500]}{'...' if len(conversation_text) > 1500 else ''}
```
### π― Recommended Actions
1. Review conversation for prospect pain points
2. Prepare personalized follow-up materials
3. Schedule discovery call within 24-48 hours
### π Lead Score: Pending Assessment
---
*This packet was auto-generated by CX AI Agent*
"""
return gr.update(visible=True, value=packet)
def escalate_to_human(chat_history):
"""Escalate conversation to human agent"""
if not chat_history:
return gr.update(visible=True, value="**π¨ Escalation Created**\n\nNo conversation history to escalate. A human agent will reach out to assist you.")
return gr.update(visible=True, value=f"""
## π¨ Escalation Created
**Status:** Pending Human Review
**Priority:** High
**Timestamp:** {datetime.now().strftime("%Y-%m-%d %H:%M")}
A human sales representative will review this conversation and reach out shortly.
**Messages in thread:** {len(chat_history)}
""")
def schedule_meeting():
"""Generate meeting scheduling info"""
from datetime import timedelta
now = datetime.now()
slots = []
for i in range(1, 4):
day = now + timedelta(days=i)
if day.weekday() < 5: # Weekdays only
slots.append(f"- {day.strftime('%A, %B %d')} at 10:00 AM EST")
slots.append(f"- {day.strftime('%A, %B %d')} at 2:00 PM EST")
return gr.update(visible=True, value=f"""
## π Meeting Scheduling
**Available Time Slots:**
{chr(10).join(slots[:4])}
To schedule a meeting, please reply with your preferred time slot, or [click here](#) to access our calendar booking system.
*Times shown in EST. Meetings are typically 30 minutes.*
""")
# Connect prospect chat handlers
prospect_send_btn.click(
fn=prospect_chat_wrapper,
inputs=[prospect_input, prospect_chatbot],
outputs=[prospect_chatbot, prospect_input]
)
prospect_input.submit(
fn=prospect_chat_wrapper,
inputs=[prospect_input, prospect_chatbot],
outputs=[prospect_chatbot, prospect_input]
)
# Connect action buttons
generate_handoff_btn.click(fn=generate_handoff_packet, inputs=[prospect_chatbot], outputs=[handoff_output])
escalate_btn.click(fn=escalate_to_human, inputs=[prospect_chatbot], outputs=[handoff_output])
schedule_btn.click(fn=schedule_meeting, outputs=[handoff_output])
return demo
if __name__ == "__main__":
demo = create_app()
demo.launch(server_name="0.0.0.0", server_port=7860, show_error=True)