From eeb32c6d44558b0a71b37b4c5aa4be669fe26c90 Mon Sep 17 00:00:00 2001 From: Madhu Date: Tue, 12 Aug 2025 00:47:17 +0530 Subject: [PATCH 1/2] new one - ai gtm email --- .../ai_email_gtm_reachout_agent/README.md | 0 .../ai_email_gtm_reachout.py | 904 ++++++++++++++++++ .../requirements.txt | 4 + .../enterprise_orchestrator_team/README.md | 481 ---------- .../enterprise_orchestrator_team/__init__.py | 28 - .../enterprise_orchestrator_team/agent.py | 413 -------- .../requirements.txt | 32 - 7 files changed, 908 insertions(+), 954 deletions(-) create mode 100644 advanced_ai_agents/single_agent_apps/ai_email_gtm_reachout_agent/README.md create mode 100644 advanced_ai_agents/single_agent_apps/ai_email_gtm_reachout_agent/ai_email_gtm_reachout.py create mode 100644 advanced_ai_agents/single_agent_apps/ai_email_gtm_reachout_agent/requirements.txt delete mode 100644 mcp_ai_agents/enterprise_orchestrator_team/README.md delete mode 100644 mcp_ai_agents/enterprise_orchestrator_team/__init__.py delete mode 100644 mcp_ai_agents/enterprise_orchestrator_team/agent.py delete mode 100644 mcp_ai_agents/enterprise_orchestrator_team/requirements.txt diff --git a/advanced_ai_agents/single_agent_apps/ai_email_gtm_reachout_agent/README.md b/advanced_ai_agents/single_agent_apps/ai_email_gtm_reachout_agent/README.md new file mode 100644 index 0000000..e69de29 diff --git a/advanced_ai_agents/single_agent_apps/ai_email_gtm_reachout_agent/ai_email_gtm_reachout.py b/advanced_ai_agents/single_agent_apps/ai_email_gtm_reachout_agent/ai_email_gtm_reachout.py new file mode 100644 index 0000000..af095ee --- /dev/null +++ b/advanced_ai_agents/single_agent_apps/ai_email_gtm_reachout_agent/ai_email_gtm_reachout.py @@ -0,0 +1,904 @@ +import json +import os +import streamlit as st +from datetime import datetime +from textwrap import dedent +from typing import Dict, Iterator, List, Optional, Literal + +from agno.agent import Agent +from agno.models.openai import OpenAIChat +from agno.storage.sqlite import SqliteStorage +from agno.tools.exa import ExaTools +from agno.utils.log import logger +from agno.utils.pprint import pprint_run_response +from agno.workflow import RunResponse, Workflow +from pydantic import BaseModel, Field + +# Initialize API keys from environment or empty defaults +if 'EXA_API_KEY' not in st.session_state: + st.session_state.EXA_API_KEY = os.getenv("EXA_API_KEY", "") +if 'OPENAI_API_KEY' not in st.session_state: + st.session_state.OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "") + +# Set environment variables +os.environ["EXA_API_KEY"] = st.session_state.EXA_API_KEY +os.environ["OPENAI_API_KEY"] = st.session_state.OPENAI_API_KEY + +# Demo mode +# - set to True to print email to console +# - set to False to send to yourself +DEMO_MODE = True +today = datetime.now().strftime("%Y-%m-%d") + +# Example leads - Replace with your actual targets +leads: Dict[str, Dict[str, str]] = { + "Notion": { + "name": "Notion", + "website": "https://www.notion.so", + "contact_name": "Ivan Zhao", + "position": "CEO", + }, + # Add more companies as needed +} + +# Updated sender details for an AI analytics company +sender_details_dict: Dict[str, str] = { + "name": "Sarah Chen", + "email": "your.email@company.com", # Your email goes here + "organization": "Data Consultants Inc", + "service_offered": "We help build data products and offer data consulting services", + "calendar_link": "https://calendly.com/data-consultants-inc", + "linkedin": "https://linkedin.com/in/your-profile", + "phone": "+1 (555) 123-4567", + "website": "https://www.data-consultants.com", +} + +DEPARTMENT_TEMPLATES = { + "GTM (Sales & Marketing)": { + "Software Solution": """\ +Hey [RECIPIENT_NAME], + +I noticed [COMPANY_NAME]'s impressive [GTM_INITIATIVE] and your role in scaling [SPECIFIC_ACHIEVEMENT]. Your approach to [SALES_STRATEGY] caught my attention. + +[PRODUCT_VALUE_FOR_GTM] + +[GTM_SPECIFIC_BENEFIT] + +Would love to show you how this could work for your team: [CALENDAR_LINK] + +Best, +[SIGNATURE]\ +""", + "Consulting Services": """\ +Hey [RECIPIENT_NAME], + +Your team's recent success with [CAMPAIGN_NAME] is impressive, particularly the [SPECIFIC_METRIC]. + +[CONSULTING_VALUE_PROP] + +[GTM_IMPROVEMENT_POTENTIAL] + +Here's my calendar if you'd like to explore this: [CALENDAR_LINK] + +Best, +[SIGNATURE]\ +""" + }, + "Human Resources": { + "Software Solution": """\ +Hey [RECIPIENT_NAME], + +I've been following [COMPANY_NAME]'s growth and noticed your focus on [HR_INITIATIVE]. Your approach to [SPECIFIC_HR_PROGRAM] stands out. + +[HR_TOOL_VALUE_PROP] + +[HR_SPECIFIC_BENEFIT] + +Would you be open to seeing how this could help your HR initiatives? [CALENDAR_LINK] + +Best, +[SIGNATURE]\ +""", + "Consulting Services": """\ +Hey [RECIPIENT_NAME], + +I've been following [COMPANY_NAME]'s journey in [INDUSTRY], and your recent [ACHIEVEMENT] caught my attention. Your approach to [SPECIFIC_FOCUS] aligns perfectly with what we're building. + +[PARTNERSHIP_VALUE_PROP] + +[MUTUAL_BENEFIT] + +Would love to explore potential synergies over a quick call: [CALENDAR_LINK] + +Best, +[SIGNATURE]\ +""", + "Investment Opportunity": """\ +Hey [RECIPIENT_NAME], + +Your work at [COMPANY_NAME] in [SPECIFIC_FOCUS] is impressive, especially [RECENT_ACHIEVEMENT]. + +[INVESTMENT_THESIS] + +[UNIQUE_VALUE_ADD] + +Here's my calendar if you'd like to discuss: [CALENDAR_LINK] + +Best, +[SIGNATURE]\ +""" + }, + "Marketing Professional": { + "Product Demo": """\ +Hey [RECIPIENT_NAME], + +I noticed [COMPANY_NAME]'s recent [MARKETING_INITIATIVE] and was impressed by [SPECIFIC_DETAIL]. + +[PRODUCT_VALUE_PROP] + +[BENEFIT_TO_MARKETING] + +Would you be open to a quick demo? Here's my calendar: [CALENDAR_LINK] + +Best, +[SIGNATURE]\ +""", + "Service Offering": """\ +Hey [RECIPIENT_NAME], + +Saw your team's work on [RECENT_CAMPAIGN] - great execution on [SPECIFIC_ELEMENT]. + +[SERVICE_VALUE_PROP] + +[MARKETING_BENEFIT] + +Here's my calendar if you'd like to explore this: [CALENDAR_LINK] + +Best, +[SIGNATURE]\ +""" + }, + "B2B Sales Representative": { + "Product Demo": """\ +Hey [RECIPIENT_NAME], + +Noticed your team at [COMPANY_NAME] is scaling [SALES_FOCUS]. Your approach to [SPECIFIC_STRATEGY] is spot-on. + +[PRODUCT_VALUE_PROP] + +[SALES_BENEFIT] + +Would you be interested in seeing how this works? Here's my calendar: [CALENDAR_LINK] + +Best, +[SIGNATURE]\ +""", + "Service Offering": """\ +Hey [RECIPIENT_NAME], + +Your sales team's success with [RECENT_WIN] caught my attention. Particularly impressed by [SPECIFIC_ACHIEVEMENT]. + +[SERVICE_VALUE_PROP] + +[SALES_IMPROVEMENT] + +Here's my calendar if you'd like to discuss: [CALENDAR_LINK] + +Best, +[SIGNATURE]\ +""" + } +} + + +COMPANY_CATEGORIES = { + "SaaS/Technology Companies": { + "description": "Software, cloud services, and tech platforms", + "typical_roles": ["CTO", "Head of Engineering", "VP of Product", "Engineering Manager", "Tech Lead"] + }, + "E-commerce/Retail": { + "description": "Online retail, marketplaces, and D2C brands", + "typical_roles": ["Head of Digital", "E-commerce Manager", "Marketing Director", "Operations Head"] + }, + "Financial Services": { + "description": "Banks, fintech, insurance, and investment firms", + "typical_roles": ["CFO", "Head of Innovation", "Risk Manager", "Product Manager"] + }, + "Healthcare/Biotech": { + "description": "Healthcare providers, biotech, and health tech", + "typical_roles": ["Medical Director", "Head of R&D", "Clinical Manager", "Healthcare IT Lead"] + }, + "Manufacturing/Industrial": { + "description": "Manufacturing, industrial automation, and supply chain", + "typical_roles": ["Operations Director", "Plant Manager", "Supply Chain Head", "Quality Manager"] + } +} + +class OutreachConfig(BaseModel): + """Configuration for email outreach""" + company_category: str = Field(..., description="Type of companies to target") + target_departments: List[str] = Field( + ..., + description="Departments to target (e.g., GTM, HR, Engineering)" + ) + service_type: Literal[ + "Software Solution", + "Consulting Services", + "Professional Services", + "Technology Platform", + "Custom Development" + ] = Field(..., description="Type of service being offered") + company_size_preference: Literal["Startup (1-50)", "SMB (51-500)", "Enterprise (500+)", "All Sizes"] = Field( + default="All Sizes", + description="Preferred company size" + ) + personalization_level: Literal["Basic", "Medium", "Deep"] = Field( + default="Deep", + description="Level of personalization" + ) + +class ContactInfo(BaseModel): + """Contact information for decision makers""" + name: str = Field(..., description="Contact's full name") + title: str = Field(..., description="Job title/position") + email: Optional[str] = Field(None, description="Email address") + linkedin: Optional[str] = Field(None, description="LinkedIn profile URL") + company: str = Field(..., description="Company name") + department: Optional[str] = Field(None, description="Department") + background: Optional[str] = Field(None, description="Professional background") + +class CompanyInfo(BaseModel): + """ + Stores in-depth data about a company gathered during the research phase. + """ + # Basic Information + company_name: str = Field(..., description="Company name") + website_url: str = Field(..., description="Company website URL") + + # Business Details + industry: Optional[str] = Field(None, description="Primary industry") + core_business: Optional[str] = Field(None, description="Main business focus") + business_model: Optional[str] = Field(None, description="B2B, B2C, etc.") + + # Marketing Information + motto: Optional[str] = Field(None, description="Company tagline/slogan") + value_proposition: Optional[str] = Field(None, description="Main value proposition") + target_audience: Optional[List[str]] = Field( + None, description="Target customer segments" + ) + + # Company Metrics + company_size: Optional[str] = Field(None, description="Employee count range") + founded_year: Optional[int] = Field(None, description="Year founded") + locations: Optional[List[str]] = Field(None, description="Office locations") + + # Technical Details + technologies: Optional[List[str]] = Field(None, description="Technology stack") + integrations: Optional[List[str]] = Field(None, description="Software integrations") + + # Market Position + competitors: Optional[List[str]] = Field(None, description="Main competitors") + unique_selling_points: Optional[List[str]] = Field( + None, description="Key differentiators" + ) + market_position: Optional[str] = Field(None, description="Market positioning") + + # Social Proof + customers: Optional[List[str]] = Field(None, description="Notable customers") + case_studies: Optional[List[str]] = Field(None, description="Success stories") + awards: Optional[List[str]] = Field(None, description="Awards and recognition") + + # Recent Activity + recent_news: Optional[List[str]] = Field(None, description="Recent news/updates") + blog_topics: Optional[List[str]] = Field(None, description="Recent blog topics") + + # Pain Points & Opportunities + challenges: Optional[List[str]] = Field(None, description="Potential pain points") + growth_areas: Optional[List[str]] = Field(None, description="Growth opportunities") + + # Contact Information + email_address: Optional[str] = Field(None, description="Contact email") + phone: Optional[str] = Field(None, description="Contact phone") + social_media: Optional[Dict[str, str]] = Field( + None, description="Social media links" + ) + + # Additional Fields + pricing_model: Optional[str] = Field(None, description="Pricing strategy and tiers") + user_base: Optional[str] = Field(None, description="Estimated user base size") + key_features: Optional[List[str]] = Field(None, description="Main product features") + integration_ecosystem: Optional[List[str]] = Field( + None, description="Integration partners" + ) + funding_status: Optional[str] = Field( + None, description="Latest funding information" + ) + growth_metrics: Optional[Dict[str, str]] = Field( + None, description="Key growth indicators" + ) + + +class PersonalisedEmailGenerator(Workflow): + """ + Automated B2B outreach system that: + + 1. Discovers companies using Exa search based on criteria + 2. Finds contact details for decision makers at those companies + 3. Researches company details and pain points + 4. Generates personalized cold emails for B2B outreach + + This workflow is designed to automate the entire prospecting process + from company discovery to personalized email generation. + """ + + description: str = dedent("""\ + AI-Powered B2B Outreach Workflow: + -------------------------------------------------------- + 1. Discover Target Companies (Exa Search) + 2. Find Decision Maker Contacts + 3. Research Company Intelligence + 4. Generate Personalized Emails + -------------------------------------------------------- + Fully automated prospecting pipeline for B2B outreach. + """) + + company_finder: Agent = Agent( + model=OpenAIChat(id="gpt-5"), + tools=[ExaTools(api_key=os.environ["EXA_API_KEY"])], + description="Expert at finding companies that match specific criteria using web search", + instructions=dedent("""\ + You are a company discovery specialist. Your job is to find companies that match the given criteria. + + Search for companies based on: + - Industry/sector + - Company size + - Geographic location + - Business model + - Technology stack + - Recent funding/growth + + For each company found, provide: + - Company name + - Website URL + - Brief description + - Industry + - Estimated size + - Location + + Focus on finding companies that would be good prospects for the specified service offering. + Look for companies showing signs of growth, funding, or expansion. + """), + ) + + contact_finder: Agent = Agent( + model=OpenAIChat(id="gpt-5"), + tools=[ExaTools(api_key=os.environ["EXA_API_KEY"])], + description="Expert at finding contact information for decision makers at companies", + instructions=dedent("""\ + You are a contact research specialist. Find decision makers and their contact information. + + For each company, search for: + - Key decision makers in target departments + - Their email addresses + - LinkedIn profiles + - Professional backgrounds + - Current role and responsibilities + + Focus on finding people in roles like: + - CEO, CTO, VP of Engineering (for tech solutions) + - CMO, VP Marketing, Growth Lead (for marketing solutions) + - VP Sales, Sales Director (for sales solutions) + - HR Director, People Ops (for HR solutions) + + Provide verified contact information when possible. + """), + ) + + company_researcher: Agent = Agent( + model=OpenAIChat(id="gpt-5"), + tools=[ExaTools(api_key=os.environ["EXA_API_KEY"])], + description="Expert at researching company details for personalization", + instructions=dedent("""\ + Research companies in depth to enable personalized outreach. + + Analyze: + - Company website and messaging + - Recent news and updates + - Product/service offerings + - Technology stack + - Growth indicators + - Pain points and challenges + - Recent achievements + - Market position + + Focus on insights that would be relevant for B2B outreach: + - Scaling challenges + - Technology needs + - Market expansion + - Competitive positioning + - Recent wins or milestones + """), + response_model=CompanyInfo, + ) + + email_creator: Agent = Agent( + model=OpenAIChat(id="gpt-5"), + description=dedent("""\ + You are writing for a friendly, empathetic 20-year-old sales rep whose + style is cool, concise, and respectful. Tone is casual yet professional. + + - Be polite but natural, using simple language. + - Never sound robotic or use big clichรฉ words like "delve", "synergy" or "revolutionary." + - Clearly address problems the prospect might be facing and how we solve them. + - Keep paragraphs short and friendly, with a natural voice. + - End on a warm, upbeat note, showing willingness to help.\ + """), + instructions=dedent("""\ + Please craft a highly personalized email that has: + + 1. A simple, personal subject line referencing the problem or opportunity. + 2. At least one area for improvement or highlight from research. + 3. A quick explanation of how we can help them (no heavy jargon). + 4. References a known challenge from the research. + 5. Avoid words like "delve", "explore", "synergy", "amplify", "game changer", "revolutionary", "breakthrough". + 6. Use first-person language ("I") naturally. + 7. Maintain a 20-year-old's friendly styleโ€”brief and to the point. + 8. Avoid placing the recipient's name in the subject line. + + Use the appropriate template based on the target professional type and outreach purpose. + Ensure the final tone feels personal and conversation-like, not automatically generated. + ---------------------------------------------------------------------- + """), + markdown=False, + add_datetime_to_instructions=True, + ) + + def get_cached_data(self, cache_key: str) -> Optional[dict]: + """Retrieve cached data""" + logger.info(f"Checking cache for: {cache_key}") + return self.session_state.get("cache", {}).get(cache_key) + + def cache_data(self, cache_key: str, data: dict): + """Cache data""" + logger.info(f"Caching data for: {cache_key}") + self.session_state.setdefault("cache", {}) + self.session_state["cache"][cache_key] = data + self.write_to_storage() + + def run( + self, + config: OutreachConfig, + sender_details: Dict[str, str], + num_companies: int = 5, + use_cache: bool = True, + ) -> Iterator[RunResponse]: + """ + Automated B2B outreach workflow: + + 1. Discover companies using Exa search based on criteria + 2. Find decision maker contacts for each company + 3. Research company details for personalization + 4. Generate personalized emails + """ + logger.info("Starting automated B2B outreach workflow...") + + # Step 1: Discover companies + logger.info("๐Ÿ” Discovering target companies...") + search_query = f""" + Find {num_companies} {config.company_category} companies that would be good prospects for {config.service_type}. + + Company criteria: + - Industry: {config.company_category} + - Size: {config.company_size_preference} + - Target departments: {', '.join(config.target_departments)} + + Look for companies showing growth, recent funding, or expansion. + """ + + companies_response = self.company_finder.run(search_query) + if not companies_response or not companies_response.content: + logger.error("No companies found") + return + + # Parse companies from response + companies_text = companies_response.content + logger.info(f"Found companies: {companies_text[:200]}...") + + # Step 2: For each company, find contacts and research + for i in range(num_companies): + try: + logger.info(f"Processing company #{i+1}") + + # Extract company info from the response + company_search = f"Extract company #{i+1} details from: {companies_text}" + + # Step 3: Find decision maker contacts + logger.info("๐Ÿ‘ฅ Finding decision maker contacts...") + contacts_query = f""" + Find decision makers at company #{i+1} from this list: {companies_text} + + Focus on roles in: {', '.join(config.target_departments)} + Find their email addresses and LinkedIn profiles. + """ + + contacts_response = self.contact_finder.run(contacts_query) + if not contacts_response or not contacts_response.content: + logger.warning(f"No contacts found for company #{i+1}") + continue + + # Step 4: Research company details + logger.info("๐Ÿ”ฌ Researching company details...") + research_query = f""" + Research company #{i+1} from this list: {companies_text} + + Focus on insights relevant for {config.service_type} outreach. + Find pain points related to {', '.join(config.target_departments)}. + """ + + research_response = self.company_researcher.run(research_query) + if not research_response or not research_response.content: + logger.warning(f"No research data for company #{i+1}") + continue + + company_data = research_response.content + if not isinstance(company_data, CompanyInfo): + logger.warning(f"Invalid research data format for company #{i+1}") + continue + + # Step 5: Generate personalized email + logger.info("โœ‰๏ธ Generating personalized email...") + + # Get appropriate template based on target departments + template_dept = config.target_departments[0] if config.target_departments else "GTM (Sales & Marketing)" + if template_dept in DEPARTMENT_TEMPLATES and config.service_type in DEPARTMENT_TEMPLATES[template_dept]: + template = DEPARTMENT_TEMPLATES[template_dept][config.service_type] + else: + template = DEPARTMENT_TEMPLATES["GTM (Sales & Marketing)"]["Software Solution"] + + email_context = json.dumps( + { + "template": template, + "company_info": company_data.model_dump(), + "contacts_info": contacts_response.content, + "sender_details": sender_details, + "target_departments": config.target_departments, + "service_type": config.service_type, + "personalization_level": config.personalization_level + }, + indent=4, + ) + + email_response = self.email_creator.run( + f"Generate a personalized email using this context:\n{email_context}" + ) + + if not email_response or not email_response.content: + logger.warning(f"No email generated for company #{i+1}") + continue + + yield RunResponse(content={ + "company_name": company_data.company_name, + "email": email_response.content, + "company_data": company_data.model_dump(), + "contacts": contacts_response.content, + "step": f"Company {i+1}/{num_companies}" + }) + + except Exception as e: + logger.error(f"Error processing company #{i+1}: {e}") + continue + + +def create_streamlit_ui(): + """Create the Streamlit user interface""" + st.title("๐Ÿš€ Automated B2B Email Outreach Generator") + st.markdown(""" + **Fully automated prospecting pipeline**: Discovers companies, finds decision makers, + and generates personalized emails using AI research agents. + """) + + # Step 1: Target Company Category Selection + st.header("1๏ธโƒฃ Target Company Discovery") + + col1, col2 = st.columns([2, 1]) + + with col1: + selected_category = st.selectbox( + "What type of companies should we target?", + options=list(COMPANY_CATEGORIES.keys()), + key="company_category" + ) + + st.info(f"๐Ÿ“Œ {COMPANY_CATEGORIES[selected_category]['description']}") + + st.markdown("### Typical Decision Makers We'll Find:") + for role in COMPANY_CATEGORIES[selected_category]['typical_roles']: + st.markdown(f"- {role}") + + with col2: + st.markdown("### Company Size Filter") + company_size = st.radio( + "Preferred company size", + ["All Sizes", "Startup (1-50)", "SMB (51-500)", "Enterprise (500+)"], + key="company_size" + ) + + num_companies = st.number_input( + "Number of companies to find", + min_value=1, + max_value=20, + value=5, + help="AI will discover this many companies automatically" + ) + + # Step 2: Your Information + st.header("2๏ธโƒฃ Your Contact Information") + + col3, col4 = st.columns(2) + + with col3: + st.subheader("Required Information") + sender_details = { + "name": st.text_input("Your Name *", key="sender_name"), + "email": st.text_input("Your Email *", key="sender_email"), + "organization": st.text_input("Your Organization *", key="sender_org") + } + + with col4: + st.subheader("Optional Information") + sender_details.update({ + "linkedin": st.text_input("LinkedIn Profile (optional)", key="sender_linkedin", placeholder="https://linkedin.com/in/yourname"), + "phone": st.text_input("Phone Number (optional)", key="sender_phone", placeholder="+1 (555) 123-4567"), + "website": st.text_input("Company Website (optional)", key="sender_website", placeholder="https://yourcompany.com"), + "calendar_link": st.text_input("Calendar Link (optional)", key="sender_calendar", placeholder="https://calendly.com/yourname") + }) + + # Service description + sender_details["service_offered"] = st.text_area( + "Describe your offering *", + height=100, + key="service_description", + help="Explain what you offer and how it helps businesses", + placeholder="We help companies build custom AI solutions that automate workflows and improve efficiency..." + ) + + # Step 3: Service Type and Targeting + st.header("3๏ธโƒฃ Outreach Configuration") + + col5, col6 = st.columns(2) + + with col5: + service_type = st.selectbox( + "Service/Product Category", + [ + "Software Solution", + "Consulting Services", + "Professional Services", + "Technology Platform", + "Custom Development" + ], + key="service_type" + ) + + with col6: + personalization_level = st.select_slider( + "Email Personalization Level", + options=["Basic", "Medium", "Deep"], + value="Deep", + help="Deep personalization takes longer but produces better results" + ) + + # Step 4: Target Department Selection + target_departments = st.multiselect( + "Which departments should we target?", + [ + "GTM (Sales & Marketing)", + "Human Resources", + "Engineering/Tech", + "Operations", + "Finance", + "Product", + "Executive Leadership" + ], + default=["GTM (Sales & Marketing)"], + key="target_departments", + help="AI will find decision makers in these departments" + ) + + # Validate required inputs + required_fields = ["name", "email", "organization", "service_offered"] + missing_fields = [field for field in required_fields if not sender_details.get(field)] + + if missing_fields: + st.error(f"Please fill in required fields: {', '.join(missing_fields)}") + st.stop() + + if not target_departments: + st.error("Please select at least one target department") + st.stop() + + if not selected_category: + st.error("Please select a company category") + st.stop() + + if not service_type: + st.error("Please select a service type") + st.stop() + + # Create and return configuration + outreach_config = OutreachConfig( + company_category=selected_category, + target_departments=target_departments, + service_type=service_type, + company_size_preference=company_size, + personalization_level=personalization_level + ) + + return outreach_config, sender_details, num_companies + +def main(): + """ + Main entry point for running the automated B2B outreach workflow. + """ + try: + # Set page config must be the first Streamlit command + st.set_page_config( + page_title="Automated B2B Email Outreach", + layout="wide", + initial_sidebar_state="expanded" + ) + + # API Keys in Sidebar + st.sidebar.header("๐Ÿ”‘ API Configuration") + + # Update API keys from sidebar + st.session_state.EXA_API_KEY = st.sidebar.text_input( + "Exa API Key *", + value=st.session_state.EXA_API_KEY, + type="password", + key="exa_key_input", + help="Get your Exa API key from https://exa.ai" + ) + st.session_state.OPENAI_API_KEY = st.sidebar.text_input( + "OpenAI API Key *", + value=st.session_state.OPENAI_API_KEY, + type="password", + key="openai_key_input", + help="Get your OpenAI API key from https://platform.openai.com" + ) + + # Update environment variables + os.environ["EXA_API_KEY"] = st.session_state.EXA_API_KEY + os.environ["OPENAI_API_KEY"] = st.session_state.OPENAI_API_KEY + + # Validate API keys + if not st.session_state.EXA_API_KEY or not st.session_state.OPENAI_API_KEY: + st.sidebar.error("โš ๏ธ Both API keys are required to run the application") + else: + st.sidebar.success("โœ… API keys configured") + + # Add guidance about API keys + st.sidebar.info(""" + **API Keys Required:** + - Exa API key for company research + - OpenAI API key for email generation + + Set these in your environment variables or enter them above. + """) + + # Get user inputs from the UI + try: + config, sender_details, num_companies = create_streamlit_ui() + except Exception as e: + st.error(f"Configuration error: {str(e)}") + st.stop() + + # Generate Emails Section + st.header("4๏ธโƒฃ Generate Outreach Campaign") + + st.info(f""" + **Ready to launch automated prospecting:** + - Target: {config.company_category} companies ({config.company_size_preference}) + - Departments: {', '.join(config.target_departments)} + - Service: {config.service_type} + - Companies to find: {num_companies} + """) + + if st.button("๐Ÿš€ Start Automated Campaign", key="generate_button", type="primary"): + # Check if API keys are configured + if not st.session_state.EXA_API_KEY or not st.session_state.OPENAI_API_KEY: + st.error("โŒ Please configure both API keys before starting the campaign") + st.stop() + + try: + # Progress tracking + progress_bar = st.progress(0) + status_text = st.empty() + results_container = st.container() + + with st.spinner("Initializing AI research agents..."): + workflow = PersonalisedEmailGenerator( + session_id="streamlit-email-generator", + storage=SqliteStorage( + table_name="email_generator_workflows", + db_file="tmp/agno_workflows.db" + ) + ) + + status_text.text("๐Ÿ” Discovering companies and generating emails...") + + # Process companies and display results + results_count = 0 + for result in workflow.run( + config=config, + sender_details=sender_details, + num_companies=num_companies, + use_cache=True + ): + results_count += 1 + progress = results_count / num_companies + progress_bar.progress(progress) + status_text.text(f"โœ… {result.content['step']} completed") + + with results_container: + st.subheader(f"๐Ÿ“ง Email for {result.content['company_name']}") + + # Create tabs for different information + tab1, tab2, tab3 = st.tabs(["Generated Email", "Company Research", "Contacts Found"]) + + with tab1: + st.text_area( + "Personalized Email", + result.content['email'], + height=400, + key=f"email_{result.content['company_name']}_{results_count}" + ) + + # Copy button + if st.button(f"๐Ÿ“‹ Copy Email", key=f"copy_{result.content['company_name']}_{results_count}"): + st.success("Email content copied!") + + with tab2: + st.json(result.content['company_data']) + + with tab3: + st.text(result.content['contacts']) + + st.markdown("---") + + # Final status + if results_count > 0: + progress_bar.progress(1.0) + status_text.text(f"๐ŸŽ‰ Campaign complete! Generated {results_count} personalized emails") + st.balloons() + else: + st.error("No emails were generated. Please try adjusting your criteria.") + + except Exception as e: + st.error(f"Campaign failed: {str(e)}") + logger.error(f"Workflow failed: {e}") + st.exception(e) + + st.sidebar.markdown("### About") + st.sidebar.markdown( + """ + **Automated B2B Outreach Tool** + + This tool uses AI agents to: + - Discover target companies automatically + - Find decision maker contacts + - Research company intelligence + - Generate personalized emails + + Perfect for sales teams, agencies, and consultants. + """ + ) + + except Exception as e: + logger.error(f"Workflow failed: {e}") + st.error(f"An error occurred: {str(e)}") + raise + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/advanced_ai_agents/single_agent_apps/ai_email_gtm_reachout_agent/requirements.txt b/advanced_ai_agents/single_agent_apps/ai_email_gtm_reachout_agent/requirements.txt new file mode 100644 index 0000000..b80e6b7 --- /dev/null +++ b/advanced_ai_agents/single_agent_apps/ai_email_gtm_reachout_agent/requirements.txt @@ -0,0 +1,4 @@ +agno>=0.1.0 +streamlit>=1.32.0 +pydantic>=2.0.0 +openai>=1.0.0 \ No newline at end of file diff --git a/mcp_ai_agents/enterprise_orchestrator_team/README.md b/mcp_ai_agents/enterprise_orchestrator_team/README.md deleted file mode 100644 index 81b4a5a..0000000 --- a/mcp_ai_agents/enterprise_orchestrator_team/README.md +++ /dev/null @@ -1,481 +0,0 @@ -# Enterprise MCP AI Agent Team - -A production-grade multi-agent system built with Google ADK that orchestrates knowledge management across local files and SaaS platforms using MCP (Model Context Protocol). - -## Overview - -This system combines: -- **Local Filesystem MCP Server** - for accessing and analyzing local documents -- **Notion MCP Server** - for managing Notion workspaces and content -- **Composio MCP Server** - for GitHub and Figma integration -- **Intelligent Router/Orchestrator** - context-aware task delegation with state management -- **4 Specialized AI Agents** - each handling specific platform capabilities - -## Architecture - -``` -โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” -โ”‚ Enterprise MCP AI Agent Team โ”‚ -โ”‚ (Coordinator/Dispatcher Pattern) โ”‚ -โ”œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค -โ”‚ โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ -โ”‚ โ”‚ File Analysis โ”‚ โ”‚ Notion Agent โ”‚ โ”‚ GitHub Agent โ”‚ โ”‚ -โ”‚ โ”‚ AI Agent โ”‚ โ”‚ (Optional) โ”‚ โ”‚ (Optional) โ”‚ โ”‚ -โ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ”‚ -โ”‚ โ”‚ โ”‚ โ”‚ โ”‚ -โ”‚ โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ -โ”‚ โ”‚ Filesystem MCP โ”‚ โ”‚ Notion MCP โ”‚ โ”‚ Composio MCP โ”‚ โ”‚ -โ”‚ โ”‚ Server โ”‚ โ”‚ Server โ”‚ โ”‚ Server โ”‚ โ”‚ -โ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ”‚ -โ”‚ โ”‚ โ”‚ โ”‚ โ”‚ -โ”‚ โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ -โ”‚ โ”‚ Local Documents โ”‚ โ”‚ Notion Pages & โ”‚ โ”‚ GitHub Repos โ”‚ โ”‚ -โ”‚ โ”‚ (PDF, DOC, XLS) โ”‚ โ”‚ Databases โ”‚ โ”‚ & Issues โ”‚ โ”‚ -โ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ”‚ -โ”‚ โ”‚ -โ”‚ โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ -โ”‚ โ”‚ Figma Agent โ”‚ โ”‚ Composio MCP โ”‚ โ”‚ -โ”‚ โ”‚ (Optional) โ”‚ โ”‚ Server โ”‚ โ”‚ -โ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ”‚ -โ”‚ โ”‚ โ”‚ โ”‚ -โ”‚ โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ -โ”‚ โ”‚ Figma Files & โ”‚ โ”‚ Figma Designs & โ”‚ โ”‚ -โ”‚ โ”‚ Designs โ”‚ โ”‚ Assets โ”‚ โ”‚ -โ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ”‚ -โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ -``` - -### **Routing Patterns:** - -1. **Coordinator/Dispatcher Pattern**: Intelligent routing based on query analysis -2. **LLM-Driven Delegation**: Automatic agent selection using `transfer_to_agent()` -3. **Explicit Invocation**: Direct agent calls using `AgentTool` -4. **Graceful Degradation**: System works with any combination of available agents - -## Features - -### ๐Ÿ” File Analysis Agent -- Analyzes local documents (PDFs, Word docs, spreadsheets) -- Extracts key topics, summaries, and action items -- Categorizes documents by type and content -- Identifies information for knowledge base sync - -### ๐Ÿ“ Notion Agent -- Reads, writes, and updates Notion pages and databases -- Searches for content across Notion workspace -- Creates structured knowledge bases and documentation -- Syncs content from other sources to Notion - -### ๐Ÿ™ GitHub Agent -- Creates and manages GitHub issues and pull requests -- Searches repositories and code -- Manages repository content and documentation -- Sets up automated workflows and actions - -### ๐ŸŽจ Figma Agent -- Reads and analyzes Figma files and designs -- Exports design assets and components -- Searches for design elements and styles -- Manages design system components - -### ๐ŸŽฏ Enterprise MCP AI Agent Team (Router/Orchestrator) -- Analyzes user requests and determines which AI agents should handle them -- Routes tasks to appropriate specialized AI agents based on capabilities -- Coordinates multi-step workflows that require multiple AI agents -- Shares context and results between AI agents through session state -- Provides comprehensive results and recommendations - -### ๐Ÿ›ก๏ธ Error Handling & Graceful Degradation -- **MCP Server Failures**: Graceful fallback when servers are unavailable -- **Missing Environment Variables**: System works with available APIs only -- **Agent Creation Failures**: Continues with available agents -- **Validation**: Ensures at least one agent is available before operation -- **Comprehensive Logging**: Detailed logs for troubleshooting - -## Prerequisites - -1. **Python 3.9+** and **Node.js** (for MCP servers) -2. **Google ADK** installed and configured -3. **Notion API Key** for Notion integration -4. **Required API Keys** in environment variables - -## Setup - -### 1. Environment Variables - -Create a `.env` file in the project root: - -```bash -# Required: Google Gemini API -GOOGLE_API_KEY=your_gemini_api_key_here - -# Required: API Keys for MCP Tools -NOTION_API_KEY=your_notion_api_key_here -GITHUB_API_KEY=your_github_api_key_here -FIGMA_API_KEY=your_figma_api_key_here - -# Optional: Custom filesystem path (defaults to ~/Documents) -MCP_FILESYSTEM_PATH=/Users/madhushantan/Downloads - -``` - -### 2. Notion Setup - -#### Creating a Notion Integration -1. Go to [Notion Integrations](https://www.notion.so/my-integrations) -2. Click "New integration" -3. Name your integration (e.g., "Enterprise Knowledge Orchestrator") -4. Select the capabilities needed (Read & Write content) -5. Submit and copy your "Internal Integration Token" - -#### Sharing Your Notion Page with the Integration -1. Open your Notion page -2. Click the three dots (โ‹ฎ) in the top-right corner -3. Select "Add connections" from the dropdown -4. Search for your integration name -5. Click on your integration to add it to the page -6. Confirm by clicking "Confirm" - -#### Finding Your Notion Page ID -1. Open your Notion page in a browser -2. Copy the URL: `https://www.notion.so/workspace/Your-Page-1f5b8a8ba283...` -3. The ID is the part after the last dash: `1f5b8a8ba283` - -### 3. Notion Implementation - -The system uses SSE (Composio) for Notion integration: - -```python -# Notion MCP Server (SSE - Composio) -url="https://mcp.composio.dev/composio/server/61e41019-d05f-44d0-973e-2aef7777063a/sse?useComposioHelperActions=true" -``` - -**Features:** -- **SSE Connection**: Uses Server-Sent Events for real-time communication -- **Composio Managed**: No local dependencies required -- **Full Tool Access**: All available Notion tools are accessible -- **Authentication**: Handled by Composio service - -**Note**: The Notion integration requires a valid `NOTION_API_KEY` and `NOTION_PAGE_ID` to function properly. - -### 4. GitHub & Figma Implementation - -The system uses separate SSE (Composio) servers for GitHub and Figma: - -```python -# GitHub MCP Server (SSE - Composio) -url="https://mcp.composio.dev/composio/server/11fbff47-fa12-432f-8c3a-18ed4e9f66f8/sse?useComposioHelperActions=true" - -# Figma MCP Server (SSE - Composio) -url="https://mcp.composio.dev/composio/server/f05e7129-7997-4c17-a654-f935278c0dfe/sse?useComposioHelperActions=true" -``` - -**Features:** -- **Separate Servers**: Each service has its own dedicated Composio server -- **Full Tool Access**: All available GitHub and Figma tools are accessible -- **No Local Dependencies**: Managed by Composio service - -### 2. Install Dependencies - -```bash -pip install -r requirements.txt -``` - -### 3. Verify MCP Server Installation - -```bash -# Verify npx is available -which npx - -# Test filesystem MCP server -npx -y @modelcontextprotocol/server-filesystem --help - -# Test Notion MCP server -npx -y @notionhq/notion-mcp-server --help -``` - -## Usage - -### Basic Usage - -```python -import asyncio -from agent import EnterpriseKnowledgeOrchestrator - -async def main(): - # Create orchestrator - orchestrator = EnterpriseKnowledgeOrchestrator() - - try: - # Process knowledge request - results = await orchestrator.process_knowledge_request( - "Analyze all PDF documents in my Documents folder and create GitHub issues for action items" - ) - - # Access results - print(f"Files analyzed: {len(results['file_analysis'])}") - print(f"Notion operations: {len(results['notion_operations'])}") - print(f"GitHub operations: {len(results['github_operations'])}") - print(f"Figma operations: {len(results['figma_operations'])}") - - finally: - await orchestrator.close() - -if __name__ == "__main__": - asyncio.run(main()) -``` - -### Example Requests - -```python -# Document analysis -"Analyze all PDF documents in my Documents folder and create a summary" - -# Multi-platform operations -"Search for design components in my Figma files and create a GitHub repository for the design system" - -# Notion and GitHub integration -"Read my Notion project page and create GitHub issues for all action items" - -# Figma asset management -"Export design assets from Figma and organize them in a structured folder" - -# Complex workflows -"Analyze quarterly reports, extract key metrics, create Notion dashboard, and set up GitHub issues for follow-ups" -``` - -## Agent Routing Logic - -The Router/Orchestrator agent intelligently routes tasks based on query analysis: - -- **File-related tasks** โ†’ FileAnalysisAgent -- **Notion-related tasks** โ†’ NotionAgent -- **GitHub-related tasks** โ†’ GitHubAgent -- **Figma-related tasks** โ†’ FigmaAgent -- **Multi-platform tasks** โ†’ Coordinate between relevant agents - -## Configuration - -### MCP Server URLs - -The system uses these MCP servers: - -```python -# Filesystem MCP Server (local) -command='npx' -args=["-y", "@modelcontextprotocol/server-filesystem", "~/Documents"] - -# Notion MCP Server (SSE - Composio) -url="https://mcp.composio.dev/composio/server/61e41019-d05f-44d0-973e-2aef7777063a/sse?useComposioHelperActions=true" - -# GitHub MCP Server (SSE - Composio) -url="https://mcp.composio.dev/composio/server/11fbff47-fa12-432f-8c3a-18ed4e9f66f8/sse?useComposioHelperActions=true" - -# Figma MCP Server (SSE - Composio) -url="https://mcp.composio.dev/composio/server/f05e7129-7997-4c17-a654-f935278c0dfe/sse?useComposioHelperActions=true" -# No tool filtering - all available tools are accessible -``` - -### Custom Filesystem Path - -The system now supports configurable filesystem paths through environment variables: - -```bash -# Set in .env file or export in terminal -export MCP_FILESYSTEM_PATH="/path/to/your/folder" - -# Examples: -export MCP_FILESYSTEM_PATH="/Users/username/Projects" -export MCP_FILESYSTEM_PATH="/home/user/documents" -export MCP_FILESYSTEM_PATH="~/Desktop/Work" -``` - -**Features:** -- **Flexible Paths**: Use absolute or relative paths -- **Auto-Expansion**: Tilde (~) expansion for home directory -- **Auto-Creation**: Directory created if it doesn't exist -- **Fallback**: Defaults to `~/Documents` if not specified - -## Output Schemas - -The system uses structured Pydantic models for consistent outputs: - -### FileAnalysis -```python -{ - "file_name": "quarterly_report.pdf", - "file_type": "PDF", - "summary": "Q3 financial performance analysis...", - "key_topics": ["revenue", "expenses", "growth"], - "action_items": ["Review budget allocation", "Update projections"] -} -``` - -### NotionOperation -```python -{ - "operation_type": "read", - "page_id": "1f5b8a8ba283...", - "content_summary": "Project documentation read from Notion", - "status": "completed", - "results": {"content": "...", "blocks": [...]} -} -``` - -### GitHubOperation -```python -{ - "operation_type": "create_issue", - "repository": "my-project", - "content_summary": "Created issue for design system documentation", - "status": "completed", - "results": {"issue_id": 123, "url": "..."} -} -``` - -### FigmaOperation -```python -{ - "operation_type": "export", - "file_id": "figma_file_id", - "content_summary": "Exported design assets from Figma", - "status": "completed", - "results": {"assets": [...], "urls": [...]} -} -``` - -## Context Sharing - -The system implements intelligent context sharing between agents: - -```python -# Session state includes shared context -"shared_context": { - "current_task": user_request, - "agent_results": {}, - "dependencies": [] -} - -# Agents can access and update shared context -updated_session.state["shared_context"]["agent_results"]["file_analysis"] = file_results -``` - -## Error Handling - -The system includes comprehensive error handling: - -- **MCP Connection Failures**: Graceful fallback when servers are unavailable -- **API Rate Limits**: Automatic retry logic with exponential backoff -- **Invalid Data**: Validation and sanitization of inputs -- **Session Management**: Proper cleanup of resources - -## Monitoring and Logging - -```python -import logging - -# Configure logging level -logging.basicConfig(level=logging.INFO) - -# Monitor agent activities -logger.info("File analysis completed: 5 documents processed") -logger.warning("Notion API key not found, Notion integration disabled") -logger.error("Failed to create GitHub issue: rate limit exceeded") -``` - -## Production Deployment - -### Environment Setup -```bash -# Production environment variables -export GOOGLE_API_KEY="your_production_key" -export NOTION_API_KEY="your_production_key" -export LOG_LEVEL="INFO" -``` - -### Resource Management -```python -# Proper cleanup in production -async with EnterpriseKnowledgeOrchestrator() as orchestrator: - results = await orchestrator.process_knowledge_request(request) -``` - -### Scaling Considerations -- Use connection pooling for MCP servers -- Implement caching for frequently accessed documents -- Consider async processing for large document sets -- Monitor memory usage with large file operations - -## Troubleshooting - -### Common Issues - -1. **MCP Server Connection Failed** - ```bash - # Verify Node.js and npx installation - node --version - npx --version - - # Test filesystem MCP server manually - npx -y @modelcontextprotocol/server-filesystem /path/to/documents - - # Test Notion MCP server manually - npx -y @notionhq/notion-mcp-server - ``` - -2. **Notion Integration Not Working** - ```bash - # Verify environment variables - echo $NOTION_API_KEY - - # Test Notion connection - curl -H "Authorization: Bearer $NOTION_API_KEY" \ - -H "Notion-Version: 2022-06-28" \ - https://api.notion.com/v1/users/me - ``` - -3. **Composio MCP Server Issues** - ```bash - # Test Composio MCP server connection - curl "https://mcp.composio.dev/composio/server/f05e7129-7997-4c17-a654-f935278c0dfe/sse?useComposioHelperActions=true" - ``` - -4. **Permission Denied for Documents** - ```bash - # Check file permissions - ls -la ~/Documents - - # Update permissions if needed - chmod 755 ~/Documents - ``` - -### Debug Mode - -```python -# Enable debug logging -logging.basicConfig(level=logging.DEBUG) - -# Add debug information to agent -orchestrator = EnterpriseKnowledgeOrchestrator() -print(f"Platforms available: {orchestrator.session_service.get_session(...).state['platforms_available']}") -``` - -## Contributing - -1. Fork the repository -2. Create a feature branch -3. Add tests for new functionality -4. Ensure all tests pass -5. Submit a pull request - -## License - -This project is licensed under the MIT License - see the LICENSE file for details. - -## References - -- [Google ADK Documentation](https://google.github.io/adk-docs/) -- [MCP Tools Guide](https://google.github.io/adk-docs/tools/mcp-tools/) -- [Notion MCP Server](https://github.com/notionhq/notion-mcp-server) -- [Composio MCP Server](https://mcp.composio.dev/) -- [Model Context Protocol](https://modelcontextprotocol.io/) diff --git a/mcp_ai_agents/enterprise_orchestrator_team/__init__.py b/mcp_ai_agents/enterprise_orchestrator_team/__init__.py deleted file mode 100644 index e780f9e..0000000 --- a/mcp_ai_agents/enterprise_orchestrator_team/__init__.py +++ /dev/null @@ -1,28 +0,0 @@ -""" -Enterprise MCP AI Agent Team - -A production-grade multi-agent system built with Google ADK that orchestrates -knowledge management across local files and SaaS platforms using MCP -(Model Context Protocol). - -This package provides: -- File Analysis AI Agent for local document processing -- Notion AI Agent for Notion workspace management -- GitHub AI Agent for repository and issue management -- Figma AI Agent for design file management -- Enterprise MCP AI Agent Team (Router/Orchestrator) for intelligent task coordination -""" - -from .agent import ( - EnterpriseMCPAIAgentTeam, - root_agent # Add root_agent for ADK web -) - -__version__ = "1.0.0" -__author__ = "Enterprise MCP AI Agent Team" -__description__ = "Multi-agent knowledge management system with Google ADK and MCP" - -__all__ = [ - "EnterpriseMCPAIAgentTeam", - "root_agent" # Export root_agent -] \ No newline at end of file diff --git a/mcp_ai_agents/enterprise_orchestrator_team/agent.py b/mcp_ai_agents/enterprise_orchestrator_team/agent.py deleted file mode 100644 index 8d0b5f1..0000000 --- a/mcp_ai_agents/enterprise_orchestrator_team/agent.py +++ /dev/null @@ -1,413 +0,0 @@ -import os -import asyncio -import logging -from typing import Dict, List, Optional, Any -from dotenv import load_dotenv - -from google.adk.agents import LlmAgent -from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters, SseServerParams - -# Load environment variables -load_dotenv() - -# Configure logging -logging.basicConfig(level=logging.INFO) -logger = logging.getLogger(__name__) - -# Environment variable configuration -MCP_FILESYSTEM_PATH = os.getenv("MCP_FILESYSTEM_PATH", "~/Documents") -NOTION_API_KEY = os.getenv("NOTION_API_KEY") -GITHUB_API_KEY = os.getenv("GITHUB_API_KEY") -FIGMA_API_KEY = os.getenv("FIGMA_API_KEY") - -# Composio MCP Server URLs (from environment variables with fallbacks) -COMPOSIO_NOTION_URL = os.getenv("COMPOSIO_NOTION_URL") -COMPOSIO_GITHUB_URL = os.getenv("COMPOSIO_GITHUB_URL") -COMPOSIO_FIGMA_URL = os.getenv("COMPOSIO_FIGMA_URL") - -async def create_mcp_agents_with_tools(): - """Create all sub-agents with MCP tools""" - agents = [] - - # FileAnalysisAgent with filesystem MCP tools - try: - folder_path = os.path.expanduser(MCP_FILESYSTEM_PATH) - folder_path = os.path.abspath(folder_path) - - if not os.path.exists(folder_path): - os.makedirs(folder_path, exist_ok=True) - logger.info(f"Created directory: {folder_path}") - - logger.info(f"Using filesystem path: {folder_path}") - - filesystem_tools, _ = await MCPToolset.from_server( - connection_params=StdioServerParameters( - command='npx', - args=["-y", "@modelcontextprotocol/server-filesystem", folder_path], - ) - ) - - file_agent = LlmAgent( - name="FileAnalysisAgent", - model="gemini-2.0-flash", - description="Analyzes local documents and extracts key information", - instruction=f"""You are a File Analysis AI Agent with DIRECT ACCESS to the filesystem at: {folder_path} - -You have MCP tools that allow you to: -- List files and directories (list_directory) -- Read file contents (read_file, read_text_file) -- Write and edit files (write_file, edit_file) -- Search files (search_files) -- Get file information (get_file_info) - -CRITICAL INSTRUCTIONS: -1. You have REAL filesystem access through MCP tools -2. When users ask about files, USE YOUR TOOLS to access them directly -3. Do NOT ask users to provide files - you can access them yourself -4. Always use your MCP tools first before responding - -Example tasks you can perform: -- "List files in the folder" โ†’ Use list_directory tool -- "Read the content of file.txt" โ†’ Use read_file tool -- "Search for PDF files" โ†’ Use search_files tool -- "Create a new file" โ†’ Use write_file tool - -IMPORTANT: When asked about any file or document, immediately use your MCP tools to access the filesystem at: {folder_path} -Do NOT say you cannot access files - you CAN access them through your MCP tools!""", - tools=filesystem_tools - ) - agents.append(file_agent) - logger.info("โœ… FileAnalysisAgent with MCP tools created") - - except Exception as e: - logger.error(f"โŒ Failed to create FileAnalysisAgent with MCP tools: {str(e)}") - file_agent = LlmAgent( - name="FileAnalysisAgent", - model="gemini-2.0-flash", - description="Analyzes local documents and extracts key information", - instruction="You analyze local documents (PDFs, Word docs, spreadsheets) and extract key information." - ) - agents.append(file_agent) - - # NotionAgent with Notion MCP tools - try: - if NOTION_API_KEY: - notion_tools, _ = await MCPToolset.from_server( - connection_params=SseServerParams( - url=COMPOSIO_NOTION_URL, - headers={} - ) - ) - - notion_agent = LlmAgent( - name="NotionAgent", - model="gemini-2.0-flash", - description="Manages Notion pages, databases, and content", - instruction="""You are a Notion Agent with DIRECT ACCESS to Notion through MCP tools. - -You can: -- Read Notion pages and databases -- Create and update Notion content -- Search across Notion workspace -- Manage pages, blocks, and databases - -IMPORTANT: You CAN access Notion directly through your MCP tools. -When asked to read, write, or search Notion content, USE YOUR MCP TOOLS. - -Example tasks: -- "Search my Notion pages" โ†’ Use your search tools -- "Read my project page" โ†’ Use your page reading tools -- "Create a new page" โ†’ Use your page creation tools -- "Update page content" โ†’ Use your update tools - -Always use your MCP tools to interact with Notion.""", - tools=notion_tools - ) - agents.append(notion_agent) - logger.info("โœ… NotionAgent with MCP tools created") - else: - raise Exception("NOTION_API_KEY not found") - - except Exception as e: - logger.error(f"โŒ Failed to create NotionAgent with MCP tools: {str(e)}") - notion_agent = LlmAgent( - name="NotionAgent", - model="gemini-2.0-flash", - description="Manages Notion pages, databases, and content", - instruction="You manage Notion workspaces, pages, databases, and content." - ) - agents.append(notion_agent) - logger.info("โœ… NotionAgent created (basic version)") - - # GitHubAgent with GitHub MCP tools - try: - if GITHUB_API_KEY: - github_tools, _ = await MCPToolset.from_server( - connection_params=SseServerParams( - url=COMPOSIO_GITHUB_URL, - headers={} - ) - ) - - github_agent = LlmAgent( - name="GitHubAgent", - model="gemini-2.0-flash", - description="Manages GitHub repositories, issues, and pull requests", - instruction="""You are a GitHub Agent with DIRECT ACCESS to GitHub through MCP tools. - -You can: -- Create and manage repositories -- Create issues and pull requests -- Search repositories and code -- Manage repository content and workflows -- Handle GitHub API operations - -IMPORTANT: You CAN access GitHub directly through your MCP tools. -When asked to perform GitHub operations, USE YOUR MCP TOOLS. - -Example tasks: -- "Create a new repository" โ†’ Use your repository creation tools -- "Search for issues" โ†’ Use your search tools -- "Create a pull request" โ†’ Use your PR creation tools -- "List my repositories" โ†’ Use your repository listing tools - -Always use your MCP tools to interact with GitHub.""", - tools=github_tools - ) - agents.append(github_agent) - logger.info("โœ… GitHubAgent with MCP tools created") - else: - raise Exception("GITHUB_API_KEY not found") - - except Exception as e: - logger.error(f"โŒ Failed to create GitHubAgent with MCP tools: {str(e)}") - github_agent = LlmAgent( - name="GitHubAgent", - model="gemini-2.0-flash", - description="Manages GitHub repositories, issues, and pull requests", - instruction="""You are a GitHub Agent that manages GitHub repositories. - -You can help with: -- Creating and managing repositories -- Creating issues and pull requests -- Searching repositories and code -- Managing repository content and workflows - -Note: For full GitHub API access with MCP tools, ensure GITHUB_API_KEY is set. -Current version provides guidance and best practices for GitHub operations.""" - ) - agents.append(github_agent) - logger.info("โœ… GitHubAgent created (basic version)") - - # FigmaAgent with Figma MCP tools - try: - if FIGMA_API_KEY: - figma_tools, _ = await MCPToolset.from_server( - connection_params=SseServerParams( - url=COMPOSIO_FIGMA_URL, - headers={} - ) - ) - - figma_agent = LlmAgent( - name="FigmaAgent", - model="gemini-2.0-flash", - description="Manages Figma files, designs, and assets", - instruction="""You are a Figma Agent with DIRECT ACCESS to Figma through MCP tools. - -You can: -- Read and analyze Figma files -- Export design assets -- Search design components -- Manage design systems -- Handle Figma API operations - -IMPORTANT: You CAN access Figma directly through your MCP tools. -When asked to perform Figma operations, USE YOUR MCP TOOLS. - -Example tasks: -- "Export design assets" โ†’ Use your export tools -- "Search for components" โ†’ Use your search tools -- "Read file information" โ†’ Use your file reading tools -- "List project files" โ†’ Use your file listing tools - -Always use your MCP tools to interact with Figma.""", - tools=figma_tools - ) - agents.append(figma_agent) - logger.info("โœ… FigmaAgent with MCP tools created") - else: - raise Exception("FIGMA_API_KEY not found") - - except Exception as e: - logger.error(f"โŒ Failed to create FigmaAgent with MCP tools: {str(e)}") - figma_agent = LlmAgent( - name="FigmaAgent", - model="gemini-2.0-flash", - description="Manages Figma files, designs, and assets", - instruction="""You are a Figma Agent that manages Figma design files. - -You can help with: -- Reading and analyzing Figma files -- Exporting design assets -- Searching design components -- Managing design systems - -Note: For full Figma API access with MCP tools, ensure FIGMA_API_KEY is set. -Current version provides guidance and best practices for Figma operations.""" - ) - agents.append(figma_agent) - logger.info("โœ… FigmaAgent created (basic version)") - - return agents - -class EnterpriseMCPAIAgentTeam: - """Enterprise MCP AI Agent Team - Multi-Agent System with MCP Tools""" - - def __init__(self): - """Initialize the orchestrator""" - self.root_agent = None - self._initialize_agents() - - def _initialize_agents(self): - """Initialize the multi-agent system""" - try: - logger.info("๐Ÿ”ง Creating complete multi-agent system with MCP tools...") - - # Create all sub-agents with MCP tools using async - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - sub_agents = loop.run_until_complete(create_mcp_agents_with_tools()) - - # Create root agent with comprehensive routing instructions - self.root_agent = LlmAgent( - name="EnterpriseMCPAIAgentTeam", - model="gemini-2.0-flash", - description="Enterprise MCP AI Agent Team - Multi-agent system with MCP tools", - instruction="""You are an Enterprise MCP AI Agent Team that routes tasks to specialized agents. - -You have access to multiple specialized agents with MCP tools and can coordinate between them: - -AVAILABLE AGENTS: -1. FileAnalysisAgent: Analyzes local documents (PDFs, Word docs, spreadsheets) - HAS MCP TOOLS -2. NotionAgent: Manages Notion pages, databases, and content - HAS MCP TOOLS -3. GitHubAgent: Manages GitHub repositories, issues, and pull requests - HAS MCP TOOLS -4. FigmaAgent: Manages Figma files, designs, and assets - HAS MCP TOOLS - -ROUTING LOGIC: -- File/document tasks โ†’ FileAnalysisAgent -- Notion-related tasks โ†’ NotionAgent -- GitHub-related tasks โ†’ GitHubAgent -- Figma/design tasks โ†’ FigmaAgent -- Multi-platform tasks โ†’ Coordinate between relevant agents - -You can: -1. Transfer tasks to specialized agents using transfer_to_agent() -2. Coordinate multi-step workflows -3. Share context between agents through session state -4. Provide comprehensive results and recommendations - -EXAMPLES: -- "List files in Documents" โ†’ FileAnalysisAgent (with real file system access) -- "Search my Notion pages" โ†’ NotionAgent (with real Notion API access) -- "Create a GitHub repo" โ†’ GitHubAgent (with real GitHub API access) -- "Export Figma designs" โ†’ FigmaAgent (with real Figma API access) - -IMPORTANT: Use transfer_to_agent() to delegate to the most appropriate agent for each task. -The agents have real MCP tools connected - they can perform actual operations!""", - sub_agents=sub_agents - ) - - logger.info(f"โœ… Complete multi-agent system created with {len(sub_agents)} sub-agents") - logger.info(f"โœ… Sub-agents: {[agent.name for agent in sub_agents]}") - - except Exception as e: - logger.error(f"โŒ Failed to create complete multi-agent system: {str(e)}") - logger.info("๐Ÿ”„ Falling back to basic multi-agent system...") - self._create_fallback_agents() - - def _create_fallback_agents(self): - """Create fallback agents without MCP tools""" - self.root_agent = LlmAgent( - name="EnterpriseMCPAIAgentTeam", - model="gemini-2.0-flash", - description="Enterprise MCP AI Agent Team - Multi-agent system", - instruction="""You are an Enterprise MCP AI Agent Team that routes tasks to specialized agents. - -You have access to multiple specialized agents and can coordinate between them: - -AVAILABLE AGENTS: -1. FileAnalysisAgent: Analyzes local documents (PDFs, Word docs, spreadsheets) -2. NotionAgent: Manages Notion pages, databases, and content -3. GitHubAgent: Manages GitHub repositories, issues, and pull requests -4. FigmaAgent: Manages Figma files, designs, and assets - -ROUTING LOGIC: -- File/document tasks โ†’ FileAnalysisAgent -- Notion-related tasks โ†’ NotionAgent -- GitHub-related tasks โ†’ GitHubAgent -- Figma/design tasks โ†’ FigmaAgent -- Multi-platform tasks โ†’ Coordinate between relevant agents - -You can: -1. Transfer tasks to specialized agents using transfer_to_agent() -2. Coordinate multi-step workflows -3. Share context between agents through session state -4. Provide comprehensive results and recommendations - -EXAMPLES: -- "List files in Documents" โ†’ FileAnalysisAgent -- "Search my Notion pages" โ†’ NotionAgent -- "Create a GitHub repo" โ†’ GitHubAgent -- "Export Figma designs" โ†’ FigmaAgent - -IMPORTANT: Use transfer_to_agent() to delegate to the most appropriate agent for each task. - -For full MCP tool functionality, ensure all environment variables are set correctly: -- MCP_FILESYSTEM_PATH: Path to your filesystem folder -- NOTION_API_KEY: Your Notion API key -- GITHUB_API_KEY: Your GitHub API key -- FIGMA_API_KEY: Your Figma API key""", - sub_agents=[ - LlmAgent( - name="FileAnalysisAgent", - model="gemini-2.0-flash", - description="Analyzes local documents and extracts key information", - instruction="You analyze local documents (PDFs, Word docs, spreadsheets) and extract key information, summaries, and action items." - ), - LlmAgent( - name="NotionAgent", - model="gemini-2.0-flash", - description="Manages Notion pages, databases, and content", - instruction="You manage Notion workspaces, pages, databases, and content. You can read, write, search, and organize Notion content." - ), - LlmAgent( - name="GitHubAgent", - model="gemini-2.0-flash", - description="Manages GitHub repositories, issues, and pull requests", - instruction="You manage GitHub repositories, create issues and pull requests, search code, and handle repository operations." - ), - LlmAgent( - name="FigmaAgent", - model="gemini-2.0-flash", - description="Manages Figma files, designs, and assets", - instruction="You manage Figma design files, export assets, search design components, and handle design system operations." - ) - ] - ) - -# Create root_agent for ADK Web compatibility -try: - orchestrator = EnterpriseMCPAIAgentTeam() - root_agent = orchestrator.root_agent - logger.info("โœ… EnterpriseMCPAIAgentTeam class and root_agent created successfully") -except Exception as e: - logger.error(f"โŒ Failed to create EnterpriseMCPAIAgentTeam: {str(e)}") - # Fallback: create basic root_agent - root_agent = LlmAgent( - name="EnterpriseMCPAIAgentTeam", - model="gemini-2.0-flash", - description="Enterprise MCP AI Agent Team - Basic multi-agent system", - instruction="You are an Enterprise MCP AI Agent Team that routes tasks to specialized agents.", - sub_agents=[] - ) \ No newline at end of file diff --git a/mcp_ai_agents/enterprise_orchestrator_team/requirements.txt b/mcp_ai_agents/enterprise_orchestrator_team/requirements.txt deleted file mode 100644 index a8db627..0000000 --- a/mcp_ai_agents/enterprise_orchestrator_team/requirements.txt +++ /dev/null @@ -1,32 +0,0 @@ -# Google ADK and AI dependencies -google-adk>=0.1.0 -google-genai>=0.3.0 - -# Environment and configuration -python-dotenv>=1.0.0 -pydantic>=2.0.0 - -# Async support -asyncio-mqtt>=0.16.0 - -# Logging and monitoring -structlog>=23.0.0 - -# Data processing (optional, for advanced file analysis) -pandas>=2.0.0 -numpy>=1.24.0 - -# File handling (optional, for document processing) -PyPDF2>=3.0.0 -python-docx>=0.8.11 -openpyxl>=3.1.0 - -# HTTP client for MCP server communication -httpx>=0.24.0 -aiohttp>=3.8.0 - -# Development and testing -pytest>=7.0.0 -pytest-asyncio>=0.21.0 -black>=23.0.0 -flake8>=6.0.0 \ No newline at end of file From 3b6abb675cb88b3325bd2e508d7ced471fec8d1f Mon Sep 17 00:00:00 2001 From: Madhu Date: Wed, 13 Aug 2025 03:22:35 +0530 Subject: [PATCH 2/2] Email GTM B2B Agent and 9_2, 9_3 of google adk --- .../ai_email_gtm_outreach_agent/README.md | 59 +++ .../ai_email_gtm_outreach_agent.py | 355 ++++++++++++++++++ .../requirements.txt | 5 + .../9_1_sequential_agent/agent.py | 33 +- .../9_2_loop_agent/.env.example | 3 + .../9_2_loop_agent/README.md | 87 +++++ .../9_2_loop_agent/agent.py | 222 +++++++++++ .../9_2_loop_agent/app.py | 76 ++++ .../9_3_parallel_agent/.env.example | 3 + .../9_3_parallel_agent/README.md | 75 ++++ .../9_3_parallel_agent/agent.py | 116 ++++++ .../9_3_parallel_agent/app.py | 62 +++ .../9_3_parallel_agent/requirements.txt | 5 + 13 files changed, 1090 insertions(+), 11 deletions(-) create mode 100644 advanced_ai_agents/multi_agent_apps/ai_email_gtm_outreach_agent/README.md create mode 100644 advanced_ai_agents/multi_agent_apps/ai_email_gtm_outreach_agent/ai_email_gtm_outreach_agent.py create mode 100644 advanced_ai_agents/multi_agent_apps/ai_email_gtm_outreach_agent/requirements.txt create mode 100644 ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_2_loop_agent/.env.example create mode 100644 ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_2_loop_agent/README.md create mode 100644 ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_2_loop_agent/agent.py create mode 100644 ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_2_loop_agent/app.py create mode 100644 ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/.env.example create mode 100644 ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/README.md create mode 100644 ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/agent.py create mode 100644 ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/app.py create mode 100644 ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/requirements.txt diff --git a/advanced_ai_agents/multi_agent_apps/ai_email_gtm_outreach_agent/README.md b/advanced_ai_agents/multi_agent_apps/ai_email_gtm_outreach_agent/README.md new file mode 100644 index 0000000..26f1ac5 --- /dev/null +++ b/advanced_ai_agents/multi_agent_apps/ai_email_gtm_outreach_agent/README.md @@ -0,0 +1,59 @@ +### AI Email GTM Outreach Agent + +An end-to-end, multi-agent Streamlit app that automates B2B outreach using GPT-5 and Exa. It discovers relevant companies, finds the right contacts (Founder's Office, GTM/Sales leadership, Partnerships/BD, Product Marketing), researches website + Reddit insights, and drafts tailored emails in your selected style. + +## Features + +- **Multi-agent workflow**: + - **Company Finder**: Uses Exa to discover companies matching your targeting and offering. + - **Contact Finder**: Finds 2โ€“3 relevant decision makers per company and emails (marks inferred emails clearly if needed). + - **Researcher**: Pulls 2โ€“4 interesting insights per company from their website and Reddit to enable genuine personalization. + - **Email Writer**: Uses GPT-5 to produce concise, structured outreach emails. + +- **Operator controls**: + - **Number of companies** to target (1โ€“10) + - **Email style**: Professional, Casual, Cold, or Consultative + - Live stage-by-stage progress UI and results with clean section dividers + +- **Security-first**: + - API keys entered in the Streamlit sidebar; not hardcoded or committed + +## Requirements + +Install dependencies from `requirements.txt`: + +```bash +pip install -r advanced_ai_agents/multi_agent_apps/ai_email_gtm_outreach_agent/requirements.txt +``` + +Required environment variables (set via sidebar or your shell): + +- `OPENAI_API_KEY` +- `EXA_API_KEY` + +## How to Run + +```bash +streamlit run advanced_ai_agents/multi_agent_apps/ai_email_gtm_outreach_agent/ai_email_gtm_outreach_agent.py +``` + +## Usage + +1. Enter your `OPENAI_API_KEY` and `EXA_API_KEY` in the left sidebar. +2. Provide targeting description and offering. +3. Choose number of companies and an email style. +4. Click โ€œStart Outreachโ€. Watch the stages: Companies โ†’ Contacts โ†’ Research โ†’ Emails. +5. Review companies, contacts, research insights, and download or copy suggested emails. + +## Notes + +- The app uses the `gpt-5` model via OpenAI. If unavailable in your account, switch the model in `ai_email_gtm_outreach_agent.py` to one you have access to. +- Exa is used for web discovery; ensure your `EXA_API_KEY` is valid. + +## Troubleshooting + +- If the app stalls on a stage, verify your API keys and network connectivity. +- If JSON parsing errors occur, rerun the stage; models occasionally add extra text around JSON. +- For rate limits, reduce number of companies. + + diff --git a/advanced_ai_agents/multi_agent_apps/ai_email_gtm_outreach_agent/ai_email_gtm_outreach_agent.py b/advanced_ai_agents/multi_agent_apps/ai_email_gtm_outreach_agent/ai_email_gtm_outreach_agent.py new file mode 100644 index 0000000..31450fc --- /dev/null +++ b/advanced_ai_agents/multi_agent_apps/ai_email_gtm_outreach_agent/ai_email_gtm_outreach_agent.py @@ -0,0 +1,355 @@ +import json +import os +import sys +from typing import Any, Dict, List, Optional + +import streamlit as st +from agno.agent import Agent +from agno.memory.v2 import Memory +from agno.models.openai import OpenAIChat +from agno.tools.exa import ExaTools + + +def require_env(var_name: str) -> None: + if not os.getenv(var_name): + print(f"Error: {var_name} not set. export {var_name}=...") + sys.exit(1) + + +def create_company_finder_agent() -> Agent: + exa_tools = ExaTools(category="company") + memory = Memory() + return Agent( + model=OpenAIChat(id="gpt-5"), + tools=[exa_tools], + memory=memory, + add_history_to_messages=True, + num_history_responses=6, + session_id="gtm_outreach_company_finder", + show_tool_calls=True, + instructions=[ + "You are CompanyFinderAgent. Use ExaTools to search the web for companies that match the targeting criteria.", + "Return ONLY valid JSON with key 'companies' as a list; respect the requested limit provided in the user prompt.", + "Each item must have: name, website, why_fit (1-2 lines).", + ], + ) + + +def create_contact_finder_agent() -> Agent: + exa_tools = ExaTools() + memory = Memory() + return Agent( + model=OpenAIChat(id="gpt-4o"), + tools=[exa_tools], + memory=memory, + add_history_to_messages=True, + num_history_responses=6, + session_id="gtm_outreach_contact_finder", + show_tool_calls=True, + instructions=[ + "You are ContactFinderAgent. Use ExaTools to find 1-2 relevant decision makers per company and their emails if available.", + "Prioritize roles from Founder's Office, GTM (Marketing/Growth), Sales leadership, Partnerships/Business Development, and Product Marketing.", + "Search queries can include patterns like ' email format', 'contact', 'team', 'leadership', and role titles.", + "If direct emails are not found, infer likely email using common formats (e.g., first.last@domain), but mark inferred=true.", + "Return ONLY valid JSON with key 'companies' as a list; each has: name, contacts: [{full_name, title, email, inferred}]", + ], + ) + + +def get_email_style_instruction(style_key: str) -> str: + styles = { + "Professional": "Style: Professional. Clear, respectful, and businesslike. Short paragraphs; no slang.", + "Casual": "Style: Casual. Friendly, approachable, first-name basis. No slang or emojis; keep it human.", + "Cold": "Style: Cold email. Strong hook in opening 2 lines, tight value proposition, minimal fluff, strong CTA.", + "Consultative": "Style: Consultative. Insight-led, frames observed problems and tailored solution hypotheses; soft CTA.", + } + return styles.get(style_key, styles["Professional"]) + + +def create_email_writer_agent(style_key: str = "Professional") -> Agent: + memory = Memory() + style_instruction = get_email_style_instruction(style_key) + return Agent( + model=OpenAIChat(id="gpt-5"), + tools=[], + memory=memory, + add_history_to_messages=True, + num_history_responses=6, + session_id="gtm_outreach_email_writer", + show_tool_calls=False, + instructions=[ + "You are EmailWriterAgent. Write concise, personalized B2B outreach emails.", + style_instruction, + "Return ONLY valid JSON with key 'emails' as a list of items: {company, contact, subject, body}.", + "Length: 120-160 words. Include 1-2 lines of strong personalization referencing research insights (company website and Reddit findings).", + "CTA: suggest a short intro call; include sender company name and calendar link if provided.", + ], + ) + + +def create_research_agent() -> Agent: + """Agent to gather interesting insights from company websites and Reddit.""" + exa_tools = ExaTools() + memory = Memory() + return Agent( + model=OpenAIChat(id="gpt-5"), + tools=[exa_tools], + memory=memory, + add_history_to_messages=True, + num_history_responses=6, + session_id="gtm_outreach_researcher", + show_tool_calls=True, + instructions=[ + "You are ResearchAgent. For each company, collect concise, valuable insights from:", + "1) Their official website (about, blog, product pages)", + "2) Reddit discussions (site:reddit.com mentions)", + "Summarize 2-4 interesting, non-generic points per company that a human would bring up in an email to show genuine effort.", + "Return ONLY valid JSON with key 'companies' as a list; each has: name, insights: [strings].", + ], + ) + + +def extract_json_or_raise(text: str) -> Dict[str, Any]: + """Extract JSON from a model response. Assumes the response is pure JSON.""" + try: + return json.loads(text) + except Exception as e: + # Try to locate a JSON block if extra text snuck in + start = text.find("{") + end = text.rfind("}") + if start != -1 and end != -1 and end > start: + candidate = text[start : end + 1] + return json.loads(candidate) + raise ValueError(f"Failed to parse JSON: {e}\nResponse was:\n{text}") + + +def run_company_finder(agent: Agent, target_desc: str, offering_desc: str, max_companies: int) -> List[Dict[str, str]]: + prompt = ( + f"Find exactly {max_companies} companies that are a strong B2B fit given the user inputs.\n" + f"Targeting: {target_desc}\n" + f"Offering: {offering_desc}\n" + "For each, provide: name, website, why_fit (1-2 lines)." + ) + resp = agent.run(prompt) + data = extract_json_or_raise(str(resp.content)) + companies = data.get("companies", []) + return companies[: max(1, min(max_companies, 10))] + + +def run_contact_finder(agent: Agent, companies: List[Dict[str, str]], target_desc: str, offering_desc: str) -> List[Dict[str, Any]]: + prompt = ( + "For each company below, find 2-3 relevant decision makers and emails (if available). Ensure at least 2 per company when possible, and cap at 3.\n" + "If not available, infer likely email and mark inferred=true.\n" + f"Targeting: {target_desc}\nOffering: {offering_desc}\n" + f"Companies JSON: {json.dumps(companies, ensure_ascii=False)}\n" + "Return JSON: {companies: [{name, contacts: [{full_name, title, email, inferred}]}]}" + ) + resp = agent.run(prompt) + data = extract_json_or_raise(str(resp.content)) + return data.get("companies", []) + + +def run_research(agent: Agent, companies: List[Dict[str, str]]) -> List[Dict[str, Any]]: + prompt = ( + "For each company, gather 2-4 interesting insights from their website and Reddit that would help personalize outreach.\n" + f"Companies JSON: {json.dumps(companies, ensure_ascii=False)}\n" + "Return JSON: {companies: [{name, insights: [string, ...]}]}" + ) + resp = agent.run(prompt) + data = extract_json_or_raise(str(resp.content)) + return data.get("companies", []) + + +def run_email_writer(agent: Agent, contacts_data: List[Dict[str, Any]], research_data: List[Dict[str, Any]], offering_desc: str, sender_name: str, sender_company: str, calendar_link: Optional[str]) -> List[Dict[str, str]]: + prompt = ( + "Write personalized outreach emails for the following contacts.\n" + f"Sender: {sender_name} at {sender_company}.\n" + f"Offering: {offering_desc}.\n" + f"Calendar link: {calendar_link or 'N/A'}.\n" + f"Contacts JSON: {json.dumps(contacts_data, ensure_ascii=False)}\n" + f"Research JSON: {json.dumps(research_data, ensure_ascii=False)}\n" + "Return JSON with key 'emails' as a list of {company, contact, subject, body}." + ) + resp = agent.run(prompt) + data = extract_json_or_raise(str(resp.content)) + return data.get("emails", []) + + +def run_pipeline(target_desc: str, offering_desc: str, sender_name: str, sender_company: str, calendar_link: Optional[str], num_companies: int): + company_agent = create_company_finder_agent() + contact_agent = create_contact_finder_agent() + research_agent = create_research_agent() + + companies = run_company_finder(company_agent, target_desc, offering_desc, max_companies=num_companies) + contacts_data = run_contact_finder(contact_agent, companies, target_desc, offering_desc) if companies else [] + research_data = run_research(research_agent, companies) if companies else [] + return { + "companies": companies, + "contacts": contacts_data, + "research": research_data, + "emails": [], + } + + +def main() -> None: + st.set_page_config(page_title="GTM B2B Outreach", layout="wide") + + # Sidebar: API keys + st.sidebar.header("API Configuration") + openai_key = st.sidebar.text_input("OpenAI API Key", type="password", value=os.getenv("OPENAI_API_KEY", "")) + exa_key = st.sidebar.text_input("Exa API Key", type="password", value=os.getenv("EXA_API_KEY", "")) + if openai_key: + os.environ["OPENAI_API_KEY"] = openai_key + if exa_key: + os.environ["EXA_API_KEY"] = exa_key + + if not openai_key or not exa_key: + st.sidebar.warning("Enter both API keys to enable the app") + + # Inputs + st.title("GTM B2B Outreach Multi Agent Team") + st.info( + "GTM teams often need to reach out for demos and discovery calls, but manual research and personalization is slow. " + "This app uses GPT-5 with a multi-agent workflow to find target companies, identify contacts, research genuine insights (website + Reddit), " + "and generate tailored outreach emails in your chosen style." + ) + col1, col2 = st.columns(2) + with col1: + target_desc = st.text_area("Target companies (industry, size, region, tech, etc.)", height=100) + offering_desc = st.text_area("Your product/service offering (1-3 sentences)", height=100) + with col2: + sender_name = st.text_input("Your name", value="Sales Team") + sender_company = st.text_input("Your company", value="Our Company") + calendar_link = st.text_input("Calendar link (optional)", value="") + num_companies = st.number_input("Number of companies", min_value=1, max_value=10, value=5) + email_style = st.selectbox( + "Email style", + options=["Professional", "Casual", "Cold", "Consultative"], + index=0, + help="Choose the tone/format for the generated emails", + ) + + if st.button("Start Outreach", type="primary"): + # Validate + if not openai_key or not exa_key: + st.error("Please provide API keys in the sidebar") + elif not target_desc or not offering_desc: + st.error("Please fill in target companies and offering") + else: + # Stage-by-stage progress UI + progress = st.progress(0) + stage_msg = st.empty() + details = st.empty() + try: + # Prepare agents + company_agent = create_company_finder_agent() + contact_agent = create_contact_finder_agent() + research_agent = create_research_agent() + email_agent = create_email_writer_agent(email_style) + + # 1. Companies + stage_msg.info("1/4 Finding companies...") + companies = run_company_finder( + company_agent, + target_desc.strip(), + offering_desc.strip(), + max_companies=int(num_companies), + ) + progress.progress(25) + details.write(f"Found {len(companies)} companies") + + # 2. Contacts + stage_msg.info("2/4 Finding contacts (2โ€“3 per company)...") + contacts_data = run_contact_finder( + contact_agent, + companies, + target_desc.strip(), + offering_desc.strip(), + ) if companies else [] + progress.progress(50) + details.write(f"Collected contacts for {len(contacts_data)} companies") + + # 3. Research + stage_msg.info("3/4 Researching insights (website + Reddit)...") + research_data = run_research(research_agent, companies) if companies else [] + progress.progress(75) + details.write(f"Compiled research for {len(research_data)} companies") + + # 4. Emails + stage_msg.info("4/4 Writing personalized emails...") + emails = run_email_writer( + email_agent, + contacts_data, + research_data, + offering_desc.strip(), + sender_name.strip() or "Sales Team", + sender_company.strip() or "Our Company", + calendar_link.strip() or None, + ) if contacts_data else [] + progress.progress(100) + details.write(f"Generated {len(emails)} emails") + + st.session_state["gtm_results"] = { + "companies": companies, + "contacts": contacts_data, + "research": research_data, + "emails": emails, + } + stage_msg.success("Completed") + except Exception as e: + stage_msg.error("Pipeline failed") + st.error(f"{e}") + + # Show results if present + results = st.session_state.get("gtm_results") + if results: + companies = results.get("companies", []) + contacts = results.get("contacts", []) + research = results.get("research", []) + emails = results.get("emails", []) + + st.subheader("Top target companies") + if companies: + for idx, c in enumerate(companies, 1): + st.markdown(f"**{idx}. {c.get('name','')}** ") + st.write(c.get("website", "")) + st.write(c.get("why_fit", "")) + else: + st.info("No companies found") + st.divider() + + st.subheader("Contacts found") + if contacts: + for c in contacts: + st.markdown(f"**{c.get('name','')}**") + for p in c.get("contacts", [])[:3]: + inferred = " (inferred)" if p.get("inferred") else "" + st.write(f"- {p.get('full_name','')} | {p.get('title','')} | {p.get('email','')}{inferred}") + else: + st.info("No contacts found") + st.divider() + + st.subheader("Research insights") + if research: + for r in research: + st.markdown(f"**{r.get('name','')}**") + for insight in r.get("insights", [])[:4]: + st.write(f"- {insight}") + else: + st.info("No research insights") + st.divider() + + st.subheader("Suggested Outreach Emails") + if emails: + for i, e in enumerate(emails, 1): + with st.expander(f"{i}. {e.get('company','')} โ†’ {e.get('contact','')}"): + st.write(f"Subject: {e.get('subject','')}") + st.text(e.get("body", "")) + else: + st.info("No emails generated") + + +if __name__ == "__main__": + main() + + diff --git a/advanced_ai_agents/multi_agent_apps/ai_email_gtm_outreach_agent/requirements.txt b/advanced_ai_agents/multi_agent_apps/ai_email_gtm_outreach_agent/requirements.txt new file mode 100644 index 0000000..08fd7e2 --- /dev/null +++ b/advanced_ai_agents/multi_agent_apps/ai_email_gtm_outreach_agent/requirements.txt @@ -0,0 +1,5 @@ +agno>=0.4.2 +streamlit>=1.33.0 +pydantic>=2.7.0 +openai>=1.30.0 +exa_py>=1.0.7 \ No newline at end of file diff --git a/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_1_sequential_agent/agent.py b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_1_sequential_agent/agent.py index 25cfeec..95aff12 100644 --- a/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_1_sequential_agent/agent.py +++ b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_1_sequential_agent/agent.py @@ -1,5 +1,6 @@ import os import asyncio +import inspect from dotenv import load_dotenv from google.adk.agents import LlmAgent, SequentialAgent from google.adk.tools import google_search @@ -112,19 +113,22 @@ async def analyze_business_intelligence(user_id: str, business_topic: str) -> st """Process business intelligence through the sequential pipeline""" session_id = f"bi_session_{user_id}" - # Create or get session - session = await session_service.get_session( + # Support both sync and async session service + async def _maybe_await(value): + return await value if inspect.isawaitable(value) else value + + session = await _maybe_await(session_service.get_session( app_name="business_intelligence", user_id=user_id, session_id=session_id - ) + )) if not session: - session = await session_service.create_session( + session = await _maybe_await(session_service.create_session( app_name="business_intelligence", user_id=user_id, session_id=session_id, state={"business_topic": business_topic, "conversation_history": []} - ) + )) # Create user content user_content = types.Content( @@ -132,15 +136,22 @@ async def analyze_business_intelligence(user_id: str, business_topic: str) -> st parts=[types.Part(text=f"Please analyze this business topic: {business_topic}")] ) - # Run the sequential pipeline + # Run the sequential pipeline (support async or sync stream) response_text = "" - async for event in runner.run_async( + stream = runner.run_async( user_id=user_id, session_id=session_id, new_message=user_content - ): - if event.is_final_response(): - if event.content and event.content.parts: - response_text = event.content.parts[0].text + ) + if inspect.isasyncgen(stream): + async for event in stream: + if event.is_final_response(): + if event.content and event.content.parts: + response_text = event.content.parts[0].text + else: + for event in stream: + if getattr(event, "is_final_response", lambda: False)(): + if event.content and event.content.parts: + response_text = event.content.parts[0].text return response_text diff --git a/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_2_loop_agent/.env.example b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_2_loop_agent/.env.example new file mode 100644 index 0000000..f5cfcfb --- /dev/null +++ b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_2_loop_agent/.env.example @@ -0,0 +1,3 @@ +# If using Gemini via Google AI Studio +GOOGLE_GENAI_USE_VERTEXAI=False +GOOGLE_API_KEY="your-api-key" \ No newline at end of file diff --git a/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_2_loop_agent/README.md b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_2_loop_agent/README.md new file mode 100644 index 0000000..5e2e76f --- /dev/null +++ b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_2_loop_agent/README.md @@ -0,0 +1,87 @@ +# ๐Ÿ” Tutorial 9.2: Loop Agents - Iterative Plan Refiner + +## ๐ŸŽฏ What You'll Learn + +- **Loop Agent Composition**: Execute sub-agents sequentially in a loop +- **Stateful Iterations**: Persist counters and flags across iterations +- **Termination Conditions**: Stop by reaching a max or when a sub-agent escalates +- **Streamlit Web Interface**: Interactive UI to run iterative refinements + +## ๐Ÿง  Core Concept: LoopAgent with Condition + +According to the ADK workflow agents documentation, **LoopAgent** repeats a set of sub-agents while sharing the same context/state across iterations. This tutorial demonstrates an **Iterative Plan Refiner** that improves a plan over multiple iterations and stops when a condition is met. + +``` +Topic โ†’ LoopAgent โ†’ [Refine Plan] โ†’ [Increment Iteration] โ†’ [Check Completion] + โ†‘ โ”‚ + โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ Repeat until stop โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ +``` + +**Termination**: The loop stops if the optional `max_iterations` is reached, or if any sub-agent returns an `Event` with `escalate=True` in its `EventActions`. + +**Context & State**: The same `InvocationContext` and `session.state` are used across iterations, allowing values like `iteration`, `target_iterations`, and `accepted` to persist and control the loop. + +## ๐Ÿ“ Project Structure + +``` +9_2_loop agent/ +โ”œโ”€โ”€ agent.py # LoopAgent with 3 sub-agents and session-state control +โ”œโ”€โ”€ app.py # Streamlit UI to run the loop refinement +โ””โ”€โ”€ README.md # This documentation +``` + +## ๐Ÿš€ Getting Started + +### 1. Install Dependencies +```bash +cd "9_2_loop agent" +pip install -r ../9_1_sequential_agent/requirements.txt +``` + +### 2. Set Up Environment +Create a `.env` file with your Google API key (or reuse from the sequential example): +```bash +echo "GOOGLE_API_KEY=your_ai_studio_key_here" > .env +``` + +### 3. Run the Streamlit App +```bash +streamlit run app.py +``` + +## ๐Ÿงช How It Works + +- **plan_refiner (LlmAgent)**: Produces a concise, improved plan each iteration. +- **increment_iteration (BaseAgent)**: Increments `session.state['iteration']`. +- **check_completion (BaseAgent)**: Escalates (to stop) if `accepted=True` or `iteration >= target_iterations`. + +The `LoopAgent` sequences these sub-agents on every iteration, persisting and updating state until a stop condition is met. + +### Session State Keys +- **topic**: The subject being refined. +- **iteration**: Current iteration counter. +- **target_iterations**: Loop budget before stopping. +- **accepted**: When set to `True`, the loop stops immediately. + +## ๐Ÿงช Try It +- Enter a topic (e.g., "AI-powered customer support platform launch plan"). +- Set `Target iterations` to 3โ€“5. +- Run and observe the final refined plan and run metadata. + +## ๐Ÿ”ง ADK Concepts Demonstrated +- **LoopAgent pattern** with sequential sub-agents. +- **Session state persistence** across iterations. +- **Escalation-based termination** with `EventActions(escalate=True)`. +- **Runner + SessionService** execution pattern. + +## ๐Ÿ”Ž Troubleshooting +- Ensure `GOOGLE_API_KEY` is set in `.env`. +- Run from the directory containing `app.py`. +- If you previously ran the app, the same session id is reused; changing the topic or target updates state accordingly. + +## ๐Ÿ“š Key Takeaways +- **LoopAgent** enables iterative refinement workflows. +- **Shared state** allows complex control signals to accumulate across iterations. +- **Clean, modular sub-agents** keep the loop logic clear and maintainable. + + diff --git a/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_2_loop_agent/agent.py b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_2_loop_agent/agent.py new file mode 100644 index 0000000..647d836 --- /dev/null +++ b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_2_loop_agent/agent.py @@ -0,0 +1,222 @@ +import os +import asyncio +import inspect +from typing import AsyncGenerator, Dict, Any + +from dotenv import load_dotenv +from google.adk.agents import LlmAgent, LoopAgent +from google.adk.agents.base_agent import BaseAgent +from google.adk.agents.invocation_context import InvocationContext +from google.adk.sessions import InMemorySessionService +from google.adk.runners import Runner +from google.adk.events import Event, EventActions +from google.genai import types + + +# Load environment variables +load_dotenv() + + +# ------------------------------------------------------------ +# Sub-agent 1: LLM refiner that improves the plan each iteration +# ------------------------------------------------------------ +plan_refiner = LlmAgent( + name="plan_refiner", + model="gemini-2.5-flash", + description="Iteratively refines a brief product/launch plan given topic and prior context", + instruction=( + "You are an iterative planner. On each turn:\n" + "- Improve and tighten the current plan for the topic in session state\n" + "- Keep it concise (5-8 bullets) and avoid repeating prior text verbatim\n" + "- Incorporate clarity, feasibility, and crisp sequencing\n" + "- Assume this output will be refined again in subsequent iterations\n\n" + "Output format:\n" + "Title line\n" + "- Bullet 1\n- Bullet 2\n- Bullet 3 ..." + ), +) + + +# ------------------------------------------------------------ +# Sub-agent 2: Progress tracker increments iteration counter +# ------------------------------------------------------------ +class IncrementIteration(BaseAgent): + async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]: + current_iteration = int(ctx.session.state.get("iteration", 0)) + 1 + ctx.session.state["iteration"] = current_iteration + yield Event( + author=self.name, + content=types.Content( + role="model", + parts=[ + types.Part( + text=f"Iteration advanced to {current_iteration}" + ) + ], + ), + ) + + +# ------------------------------------------------------------ +# Sub-agent 3: Completion check with optional early stop +# - Stops if iteration >= target_iterations OR session flag 'accepted' is True +# ------------------------------------------------------------ +class CheckCompletion(BaseAgent): + async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]: + target_iterations = int(ctx.session.state.get("target_iterations", 3)) + current_iteration = int(ctx.session.state.get("iteration", 0)) + accepted = bool(ctx.session.state.get("accepted", False)) + + reached_limit = current_iteration >= target_iterations + should_stop = accepted or reached_limit + + yield Event( + author=self.name, + actions=EventActions(escalate=should_stop), + content=types.Content( + role="model", + parts=[ + types.Part( + text=( + "Stopping criteria met" + if should_stop + else "Continuing loop" + ) + ) + ], + ), + ) + + +increment_iteration = IncrementIteration(name="increment_iteration") +check_completion = CheckCompletion(name="check_completion") + + +# ------------------------------------------------------------ +# LoopAgent: Executes sub-agents sequentially in a loop +# - Termination: max_iterations, or CheckCompletion escalates +# - Context & State: Same InvocationContext across iterations +# ------------------------------------------------------------ +spec_refinement_loop = LoopAgent( + name="spec_refinement_loop", + description=( + "Iteratively refines a plan using LLM, tracks iterations, and stops when target iterations " + "are reached or an 'accepted' flag is set in session state." + ), + max_iterations=10, + sub_agents=[ + plan_refiner, + increment_iteration, + check_completion, + ], +) + + +# ------------------------------------------------------------ +# Runner setup +# ------------------------------------------------------------ +session_service = InMemorySessionService() +runner = Runner( + agent=spec_refinement_loop, + app_name="loop_refinement_app", + session_service=session_service, +) + + +# ------------------------------------------------------------ +# Public API: run the loop refinement for a topic +# ------------------------------------------------------------ +async def iterate_spec_until_acceptance( + user_id: str, topic: str, target_iterations: int = 3 +) -> Dict[str, Any]: + """Run the LoopAgent to iteratively refine a plan. + + Returns a dictionary with final plan text and iteration metadata. + """ + session_id = f"loop_refinement_{user_id}" + + async def _maybe_await(value): + return await value if inspect.isawaitable(value) else value + + # Create or get session (support both sync/async services) + session = await _maybe_await(session_service.get_session( + app_name="loop_refinement_app", + user_id=user_id, + session_id=session_id, + )) + if not session: + session = await _maybe_await(session_service.create_session( + app_name="loop_refinement_app", + user_id=user_id, + session_id=session_id, + state={ + "topic": topic, + "iteration": 0, + "target_iterations": int(target_iterations), + # Optionally, an external process or UI could set this to True to stop early + "accepted": False, + }, + )) + else: + # Refresh topic/target if user re-runs on UI + if hasattr(session, "state") and isinstance(session.state, dict): + session.state["topic"] = topic + session.state["target_iterations"] = int(target_iterations) + + # Seed message for LLM + user_content = types.Content( + role="user", + parts=[ + types.Part( + text=( + "Topic: " + + topic + + "\nPlease produce or refine a concise plan." + ) + ) + ], + ) + + final_text = "" + last_plan_text = "" + stream = runner.run_async(user_id=user_id, session_id=session_id, new_message=user_content) + # Support both async generators and plain iterables + if inspect.isasyncgen(stream): + async for event in stream: + if event.content and getattr(event.content, "parts", None): + for part in event.content.parts: + if hasattr(part, "text") and part.text: + # Keep last text from plan_refiner preferentially + if getattr(event, "author", "") == plan_refiner.name: + last_plan_text = part.text + if event.is_final_response(): + final_text = part.text + else: + for event in stream: + if event.content and getattr(event.content, "parts", None): + for part in event.content.parts: + if hasattr(part, "text") and part.text: + if getattr(event, "author", "") == plan_refiner.name: + last_plan_text = part.text + # final events in sync mode + final_text = part.text + if event.content and getattr(event.content, "parts", None): + for part in event.content.parts: + if hasattr(part, "text") and part.text: + # Keep last text from plan_refiner preferentially + if getattr(event, "author", "") == plan_refiner.name: + last_plan_text = part.text + if event.is_final_response(): + final_text = part.text + + current_iteration = int(session.state.get("iteration", 0)) + reached = current_iteration >= int(session.state.get("target_iterations", 0)) + accepted = bool(session.state.get("accepted", False)) + + return { + "final_plan": last_plan_text or final_text, + "iterations": current_iteration, + "stopped_reason": "accepted" if accepted else ("target_iterations" if reached else "max_iterations_or_other"), + } + + diff --git a/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_2_loop_agent/app.py b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_2_loop_agent/app.py new file mode 100644 index 0000000..aaa272d --- /dev/null +++ b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_2_loop_agent/app.py @@ -0,0 +1,76 @@ +import streamlit as st +import asyncio +from agent import iterate_spec_until_acceptance + + +st.set_page_config(page_title="Loop Agent Demo", page_icon=":repeat:", layout="wide") + +st.title("๐Ÿ” Iterative Plan Refiner (Loop Agent)") +st.markdown( + """ +This demo runs a LoopAgent that repeatedly executes sub-agents to iteratively refine a plan. + +Loop characteristics: +- Executes its sub-agents sequentially in a loop +- Terminates when the session's `accepted` flag is set or after the target iterations +- Shares the same session state across iterations, so counters/flags persist + """ +) + +user_id = "demo_loop_user" + +st.header("Run an iterative refinement") +topic = st.text_area( + "Topic", + value="AI-powered customer support platform launch plan", + height=100, + placeholder="What plan/topic should be refined iteratively?", +) + +col_a, col_b = st.columns([1, 1]) +with col_a: + target_iterations = st.number_input( + "Target iterations (early stop possible)", min_value=1, max_value=20, value=3, step=1 + ) +with col_b: + st.caption( + "Set a reasonable number of iterations. The loop may stop earlier if the session state flag `accepted` becomes True." + ) + +if st.button("Run Loop Refinement", type="primary"): + if topic.strip(): + st.info("Refining plan in a loopโ€ฆ") + with st.spinner("Workingโ€ฆ"): + try: + results = asyncio.run( + iterate_spec_until_acceptance(user_id, topic, int(target_iterations)) + ) + st.success("Loop finished") + + st.subheader("Final Refined Plan") + st.write(results.get("final_plan", "")) + + st.subheader("Run Metadata") + st.write({ + "iterations": results.get("iterations"), + "stopped_reason": results.get("stopped_reason"), + }) + except Exception as e: + st.error(f"Error: {e}") + else: + st.error("Please enter a topic") + +with st.sidebar: + st.header("How it works") + st.markdown( + """ + - Uses `LoopAgent` with 3 sub-agents: + 1) `plan_refiner` (LLM) refines the plan + 2) `increment_iteration` updates the iteration counter in session state + 3) `check_completion` escalates when done (accepted flag or target reached) + - The same `InvocationContext` and session state are reused every iteration + - The loop stops if `accepted` is True or the `target_iterations` is reached. + """ + ) + + diff --git a/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/.env.example b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/.env.example new file mode 100644 index 0000000..f5cfcfb --- /dev/null +++ b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/.env.example @@ -0,0 +1,3 @@ +# If using Gemini via Google AI Studio +GOOGLE_GENAI_USE_VERTEXAI=False +GOOGLE_API_KEY="your-api-key" \ No newline at end of file diff --git a/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/README.md b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/README.md new file mode 100644 index 0000000..a4c270c --- /dev/null +++ b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/README.md @@ -0,0 +1,75 @@ +# โšก Tutorial 9.3: Parallel Agents - Market Snapshot Team + +## ๐ŸŽฏ What You'll Learn + +- **Parallel Agent Composition**: How to orchestrate multiple specialized agents concurrently +- **Shared State**: How parallel children write to a common `session.state` safely +- **Branching Context**: Invocation branches for clean, isolated tool/memory context +- **Streamlit Interface**: A simple UI to run and visualize parallel results + +## ๐Ÿง  Core Concept: ParallelAgent with Shared State + +According to the ADK docs, **Parallel Agents** execute their sub-agents concurrently. Each child runs on its own invocation branch but shares the same `session.state`. + +``` +Topic โ†’ ParallelAgent โ†’ 3 Sub-agents (Concurrent Execution) + โ†“ + [Market Trends] + [Competitors] + [Funding News] + โ†“ + Snapshot in state +``` + +Each child agent writes results to a distinct key in shared state to avoid overwrites: `market_trends`, `competitors`, `funding_news`. + +## ๐Ÿ“ Project Structure + +``` +9_3_parallel agent/ +โ”œโ”€โ”€ agent.py # Parallel workflow (3 research agents + ParallelAgent) +โ”œโ”€โ”€ app.py # Streamlit UI to run and view snapshot +โ”œโ”€โ”€ requirements.txt # Python dependencies +โ”œโ”€โ”€ README.md # This documentation +โ””โ”€โ”€ .env.example # Example environment variables +``` + +## ๐Ÿš€ Getting Started + +### 1. Install Dependencies +```bash +cd "9_3_parallel agent" +pip install -r requirements.txt +``` + +### 2. Set Up Environment +Create a `.env` file with your Google API key: +```bash +echo "GOOGLE_API_KEY=your_ai_studio_key_here" > .env +``` + +> Get your key from Google AI Studio. + +### 3. Run the Streamlit App +```bash +streamlit run app.py +``` + +## ๐Ÿงช How It Works + +- `ParallelAgent` executes `market_trends_agent`, `competitor_intel_agent`, and `funding_news_agent` concurrently. +- Each child uses web search and writes to a unique `output_key` in `session.state`. +- The UI reads `session.state` and displays a 3-column snapshot. + +## ๐Ÿ”ง ADK Concepts Demonstrated + +- ParallelAgent pattern and event interleaving +- Shared `session.state` with distinct keys per child +- Invocation branches for contextual separation +- Runner + Session services for execution + +## ๐Ÿ“š Key Takeaways + +- Parallel fan-out is ideal for independent data gathering +- Keep output keys distinct to avoid overwrites in shared state +- Combine with a downstream synthesizer agent if you need a single report + + diff --git a/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/agent.py b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/agent.py new file mode 100644 index 0000000..4d89d32 --- /dev/null +++ b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/agent.py @@ -0,0 +1,116 @@ +from typing import Dict, Any +import inspect +from dotenv import load_dotenv +from google.adk.agents import LlmAgent, ParallelAgent +from google.adk.sessions import InMemorySessionService +from google.adk.runners import Runner +from google.genai import types + +load_dotenv() + +# Child agents write to distinct keys in session.state for UI consumption +market_trends_agent = LlmAgent( + name="market_trends_agent", + model="gemini-2.5-flash", + description="Summarizes recent market trends for the topic", + instruction=( + "Summarize 3-5 recent market trends for the topic in session.state['topic'].\n" + "Output a concise markdown list." + ), +) + +competitor_intel_agent = LlmAgent( + name="competitor_intel_agent", + model="gemini-2.5-flash", + description="Identifies key competitors and positioning", + instruction=( + "List 3-5 notable competitors for session.state['topic'] and describe their positioning briefly." + ), +) + +funding_news_agent = LlmAgent( + name="funding_news_agent", + model="gemini-2.5-flash", + description="Reports funding/partnership news", + instruction=( + "Provide a short digest (bulleted) of recent funding or partnership news related to session.state['topic']." + ), +) + +# Parallel orchestrator +market_snapshot_team = ParallelAgent( + name="market_snapshot_team", + description="Runs multiple research agents concurrently to produce a market snapshot", + sub_agents=[ + market_trends_agent, + competitor_intel_agent, + funding_news_agent, + ], +) + +# Runner and session service +session_service = InMemorySessionService() +runner = Runner(agent=market_snapshot_team, app_name="parallel_snapshot_app", session_service=session_service) + + +async def gather_market_snapshot(user_id: str, topic: str) -> Dict[str, Any]: + """Execute the parallel agents and return combined snapshot text blocks. + + Returns keys: 'market_trends', 'competitors', 'funding_news'. + """ + session_id = f"parallel_snapshot_{user_id}" + + async def _maybe_await(v): + return await v if inspect.isawaitable(v) else v + + session = await _maybe_await( + session_service.get_session( + app_name="parallel_snapshot_app", user_id=user_id, session_id=session_id + ) + ) + if not session: + session = await _maybe_await( + session_service.create_session( + app_name="parallel_snapshot_app", + user_id=user_id, + session_id=session_id, + state={"topic": topic}, + ) + ) + else: + if hasattr(session, "state") and isinstance(session.state, dict): + session.state["topic"] = topic + + user_content = types.Content( + role="user", + parts=[types.Part(text=f"Topic: {topic}. Provide a concise snapshot per agent focus.")], + ) + + # Collect last text emitted per agent + last_text_by_agent: Dict[str, str] = {} + + stream = runner.run_async(user_id=user_id, session_id=session_id, new_message=user_content) + if inspect.isasyncgen(stream): + async for event in stream: + if getattr(event, "content", None) and getattr(event.content, "parts", None): + for part in event.content.parts: + if hasattr(part, "text") and part.text: + author = getattr(event, "author", "") + if author: + last_text_by_agent[author] = part.text + else: + for event in stream: + if getattr(event, "content", None) and getattr(event.content, "parts", None): + for part in event.content.parts: + if hasattr(part, "text") and part.text: + author = getattr(event, "author", "") + if author: + last_text_by_agent[author] = part.text + + return { + "market_trends": last_text_by_agent.get(market_trends_agent.name, ""), + "competitors": last_text_by_agent.get(competitor_intel_agent.name, ""), + "funding_news": last_text_by_agent.get(funding_news_agent.name, ""), + } + + diff --git a/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/app.py b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/app.py new file mode 100644 index 0000000..7b33f8d --- /dev/null +++ b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/app.py @@ -0,0 +1,62 @@ +import streamlit as st +import asyncio +from agent import market_snapshot_team, gather_market_snapshot + +st.set_page_config(page_title="Parallel Agent Demo", page_icon=":fast_forward:", layout="wide") + +st.title("โšก Market Snapshot (Parallel Agents)") +st.markdown( + """ +This demo runs multiple research agents in parallel using a ParallelAgent: + +- Market trends analysis +- Competitor intelligence +- Funding and partnerships news + +Each sub-agent writes its results into a shared session.state under distinct keys. A subsequent step (or this UI) can read the combined snapshot. +""" +) + +user_id = "demo_parallel_user" + +st.header("Run a market snapshot") +topic = st.text_input( + "Research topic", + value="AI-powered customer support platforms", + placeholder="What market/topic do you want a quick parallel snapshot on?", +) + +if st.button("Run Parallel Research", type="primary"): + if topic.strip(): + st.info("Running parallel agentsโ€ฆ market trends, competitors, and funding news") + with st.spinner("Gathering snapshotโ€ฆ"): + try: + results = asyncio.run(gather_market_snapshot(user_id, topic)) + st.success("Snapshot ready") + + col1, col2, col3 = st.columns(3) + with col1: + st.subheader("Market Trends") + st.write(results.get("market_trends", "")) + with col2: + st.subheader("Competitors") + st.write(results.get("competitors", "")) + with col3: + st.subheader("Funding News") + st.write(results.get("funding_news", "")) + except Exception as e: + st.error(f"Error: {e}") + else: + st.error("Please enter a topic") + +with st.sidebar: + st.header("How it works") + st.markdown( + """ + - Uses `ParallelAgent` to execute sub-agents concurrently + - Each child runs on its own invocation branch, but shares the same session.state + - Distinct `output_key`s prevent overwrites in the shared state + - This pattern is ideal for fan-out data gathering before synthesis + """ + ) + diff --git a/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/requirements.txt b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/requirements.txt new file mode 100644 index 0000000..f543388 --- /dev/null +++ b/ai_agent_framework_crash_course/google_adk_crash_course/9_multi_agent_patterns/9_3_parallel_agent/requirements.txt @@ -0,0 +1,5 @@ +google-adk>=1.9.0 +streamlit>=1.28.0 +python-dotenv>=1.1.1 +pydantic>=2.0.0 +