Merge pull request #308 from Madhuvod/gpt-mcp-agent

Added new demo

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.
+
+

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 '<Company> 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()
+
+

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

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()

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

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

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"

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.
+
+

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"),
+    }
+
+

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.
+        """
+    )
+
+

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"

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
+
+

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, ""),
+    }
+
+

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
+        """
+    )
+

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
+