How to Create a Multi-Domain AI-Web Agent using Notte and Gemini

Google-generativeai request beautifulsoup4
Install chromium with a patchright.


import os
Import json
import time
Type import list, Optional
BaseModel pydantic imported
import google.generativeai as genai
Dotenv Import Load_Dotenv


GEMINI_API_KEY = "USE YOUR OWN API KEY HERE"
os.environ['GEMINI_API_KEY'] = GEMINI_API_KEY
genai.configure(api_key=GEMINI_API_KEY)


Import Notte

class ProductInfo(BaseModel):
 Name:
 Price: str
 Rating Optional[float]
   availability: str
   description: str


class NewsArticle(BaseModel):
 Title:
 Summary
 The URL is:
 Date:
   source: str


class SocialMediaPost(BaseModel):
 Content:
   author: str
 Int
   timestamp: str
 Platform: str


class SearchResult(BaseModel):
   query: str
 Results: List[dict]
   total_found: int

Class AdvancedNotteAgent
   def __init__(self, headless=True, max_steps=20):
 "Self-headless" = "headless
       self.max_steps = max_steps
 Self-session = None
 Self-agent = None
      
   def __enter__(self):
       self.session = notte.Session(headless=self.headless)
       self.session.__enter__()
       self.agent = notte.Agent(
           session=self.session,
           reasoning_model="gemini/gemini-2.5-flash",
           max_steps=self.max_steps
       )
 Return self
      
   def __exit__(self, exc_type, exc_val, exc_tb):
 If self.session
           self.session.__exit__(exc_type, exc_val, exc_tb)


   def research_product(self, product_name: str, website: str = "amazon.com") -> ProductInfo:
       """Research a product and extract structured information"""
 Task = F"Go to {website}, search for '{product_name}', click on the first relevant product, and extract detailed product information including name, price, rating, availability, and description."
      
       response = self.agent.run(
           task=task,
           response_format=ProductInfo,
           url=f"https://{website}"
       )
 Answer back


   def news_aggregator(self, topic: str, num_articles: int = 3) -> List[NewsArticle]:
       """Aggregate news articles on a specific topic"""
 Simply put, task = "f""Search for recent news about '{topic}', find {num_articles} relevant articles, and extract title, summary, URL, date, and source for each."
      
       response = self.agent.run(
           task=task,
           url="https://news.google.com",
           response_format=List[NewsArticle]
       )
 Answer back


   def social_media_monitor(self, hashtag: str, platform: str = "twitter") -> List[SocialMediaPost]:
       """Monitor social media for specific hashtags"""
 Lower the platform if it is lower() == "twitter":
 http://url= "https://twitter.com"
 Elif lower platform() == "reddit":
 http://url= "https://reddit.com"
       else:
 URL = f"https://{platform}.com"
          
 Task = f"Go to {platform}, search for posts with hashtag '{hashtag}', and extract content, author, engagement metrics, and timestamps from the top 5 posts."
      
       response = self.agent.run(
           task=task,
           url=url,
           response_format=List[SocialMediaPost]
       )
 Answer back


   def competitive_analysis(self, company: str, competitors: List[str]) -> dict:
       """Perform competitive analysis by gathering pricing and feature data"""
       results = {}
      
 Competitor in [company] + competitors:
 Task = f"Go to {competitor}'s website, find their pricing page or main product page, and extract key features, pricing tiers, and unique selling points."
          
           try:
               response = self.agent.run(
                   task=task,
                   url=f"https://{competitor}.com"
               )
 The following are results of the search:[competitor] = response.answer
               time.sleep(2) 
 Exception to the rule:
 The following are results of the search:[competitor] = f"Error: {str(e)}"
              
 Results of the return


   def job_market_scanner(self, job_title: str, location: str = "remote") -> List[dict]:
       """Scan job market for opportunities"""
 Task = F"Search for '{job_title}' jobs in '{location}', extract job titles, companies, salary ranges, and application URLs from the first 10 results."
      
       response = self.agent.run(
           task=task,
           url="https://indeed.com"
       )
 Answer back


   def price_comparison(self, product: str, websites: List[str]) -> dict:
       """Compare prices across multiple websites"""
       price_data = {}
      
 Sites for websites
 If you want to know what task is, just type f."Search for '{product}' on this website and find the best price, including any discounts or special offers."
          
           try:
               response = self.agent.run(
                   task=task,
                   url=f"https://{site}"
               )
               price_data[site] = response.answer
               time.sleep(1)
 “except Exception” is a phrase that means:
               price_data[site] = f"Error: {str(e)}"
              
 Return Price Data


   def content_research(self, topic: str, content_type: str = "blog") -> dict:
       """Research content ideas and trending topics"""
 If content_type== "blog":
 http://url= "https://medium.com"
 Task = F"Search for '{topic}' articles, analyze trending content, and identify popular themes, engagement patterns, and content gaps."
 If content_type is equal to, then the following will be displayed: "video":
 http://url= "https://youtube.com"
 Task = F"Search for '{topic}' videos, analyze view counts, titles, and descriptions to identify trending formats and popular angles."
       else:
 http://url= "https://google.com"
 task = fReturn"Search for '{topic}' content across the web and analyze trending discussions and popular formats."
          
       response = self.agent.run(task=task, url=url)
       return {"topic": topic, "insights": response.answer, "platform": content_type}

def demo_ecommerce_research():
   """Demo: E-commerce product research and comparison"""
   print("🛍️ E-commerce Research Demo")
   print("=" * 50)
  
   with AdvancedNotteAgent(headless=True) as agent:
       product = agent.research_product("wireless earbuds", "amazon.com")
       print(f"Product Research Results:")
       print(f"Name: {product.name}")
       print(f"Price: {product.price}")
       print(f"Rating: {product.rating}")
       print(f"Availability: {product.availability}")
       print(f"Description: {product.description[:100]}...")
      
       print("n💰 Price Comparison:")
 Sites ["amazon.com", "ebay.com", "walmart.com"]
       prices = agent.price_comparison("wireless earbuds", websites)
 Site data, prices and items():
           print(f"{site}: {data}")


def demo_news_intelligence():
   """Demo: News aggregation and analysis"""
   print("📰 News Intelligence Demo")
   print("=" * 50)
  
 AdvancedNotteAgent() As agent
       articles = agent.news_aggregator("artificial intelligence", 3)
      
 For i, enumerate the articles (1, 2):
           print(f"nArticle {i}:")
           print(f"Title: {article.title}")
           print(f"Source: {article.source}")
           print(f"Summary: {article.summary}")
           print(f"URL: {article.url}")


def demo_social_listening():
   """Demo: Social media monitoring and sentiment analysis"""
   print("👂 Social Media Listening Demo")
   print("=" * 50)
  
 AdvancedNotteAgent() As agent
       posts = agent.social_media_monitor("#AI", "reddit")
      
 For i, enter the number of posts in (1):
           print(f"nPost {i}:")
           print(f"Author: {post.author}")
           print(f"Content: {post.content[:100]}...")
           print(f"Engagement: {post.likes} likes")
           print(f"Platform: {post.platform}")


def demo_market_intelligence():
   """Demo: Competitive analysis and market research"""
   print("📊 Market Intelligence Demo")
   print("=" * 50)
  
 AdvancedNotteAgent() As agent
 That's a lot of company. "openai"
 Competitors = ["anthropic", "google"]
       analysis = agent.competitive_analysis(company, competitors)
      
 Compilation of data for analysis.():
           print(f"n{comp.upper()}:")
           print(f"Analysis: {str(data)[:200]}...")


def demo_job_market_analysis():
   """Demo: Job market scanning and analysis"""
   print("💼 Job Market Analysis Demo")
   print("=" * 50)
  
 AdvancedNotteAgent() As agent
       jobs = agent.job_market_scanner("python developer", "san francisco")
      
       print(f"Found {len(jobs)} job opportunities:")
 Job in Jobs[:3]:  
           print(f"- {job}")


def demo_content_strategy():
   """Demo: Content research and trend analysis"""
   print("✍️ Content Strategy Demo")
   print("=" * 50)
  
 AdvancedNotteAgent() As agent
       blog_research = agent.content_research("machine learning", "blog")
       video_research = agent.content_research("machine learning", "video")
      
       print("Blog Content Insights:")
       print(blog_research["insights"][:300] + "...")
      
       print("nVideo Content Insights:")
       print(video_research["insights"][:300] + "...")

If you want to classify WorkflowManager, then do so.
   def __init__(self):
       self.agents = []
       self.results = {}
      
   def add_agent_task(self, name: str, task_func, *args, **kwargs):
       """Add an agent task to the workflow"""
       self.agents.append({
           'name': name,
           'func': task_func,
           'args': args,
 "Kwargs:"
       })
      
   def execute_workflow(self, parallel=False):
       """Execute all agent tasks in the workflow"""
       print("🚀 Executing Multi-Agent Workflow")
       print("=" * 50)
      
 Agent_task is a self-agent task.
 name = agent_task['name']
           func = agent_task['func']
 Agent_task = args['args']
 Agent_task = Kwargs['kwargs']
          
           print(f"n🤖 Executing {name}...")
           try:
               result = func(*args, **kwargs)
               self.results[name] Enjoy the result
               print(f"✅ {name} completed successfully")
 "except Exception" is a phrase that means:
               self.results[name] This is a f"Error: {str(e)}"
               print(f"❌ {name} failed: {str(e)}")
              
 If not parallel
               time.sleep(2) 
 Return self.results


def market_research_workflow(company_name: str, product_category: str):
   """Complete market research workflow"""
 Workflow Manager = workflow()
  
   workflow.add_agent_task(
       "Product Research",
       lambda: research_trending_products(product_category)
   )
  
   workflow.add_agent_task(
       "Competitive Analysis",
       lambda: analyze_competitors(company_name, product_category)
   )
  
   workflow.add_agent_task(
       "Social Sentiment",
       lambda: monitor_brand_sentiment(company_name)
   )
  
   return workflow.execute_workflow()


def research_trending_products(category: str):
   """Research trending products in a category"""
   with AdvancedNotteAgent(headless=True) as agent:
 Task = f"Research trending {category} products, find top 5 products with prices, ratings, and key features."
       response = agent.agent.run(
           task=task,
           url="https://amazon.com"
       )
 Answer back


def analyze_competitors(company: str, category: str):
   """Analyze competitors in the market"""
   with AdvancedNotteAgent(headless=True) as agent:
 Task = f"Research {company} competitors in {category}, compare pricing strategies, features, and market positioning."
       response = agent.agent.run(
           task=task,
           url="https://google.com"
       )
 Answer back


def monitor_brand_sentiment(brand: str):
   """Monitor brand sentiment across platforms"""
   with AdvancedNotteAgent(headless=True) as agent:
 Task = f"Search for recent mentions of {brand} on social media and news, analyze sentiment and key themes."
       response = agent.agent.run(
           task=task,
           url="https://reddit.com"
       )
 Answer back

Def main():
   """Main function to run all demos"""
   print("🚀 Advanced Notte AI Agent Tutorial")
   print("=" * 60)
   print("Note: Make sure to set your GEMINI_API_KEY above!")
   print("Get your free API key at: https://makersuite.google.com/app/apikey")
   print("=" * 60)
  
 if GEMINI_API_KEY = "YOUR_GEMINI_API_KEY":
       print("❌ Please set your GEMINI_API_KEY in the code above!")
 You can return to your original language by clicking here.
  
   try:
       print("n1. E-commerce Research Demo")
       demo_ecommerce_research()
      
       print("n2. News Intelligence Demo")
       demo_news_intelligence()
      
       print("n3. Social Media Listening Demo")
       demo_social_listening()
      
       print("n4. Market Intelligence Demo")
       demo_market_intelligence()
      
       print("n5. Job Market Analysis Demo")
       demo_job_market_analysis()
      
       print("n6. Content Strategy Demo")
       demo_content_strategy()
      
       print("n7. Multi-Agent Workflow Demo")
       results = market_research_workflow("Tesla", "electric vehicles")
       print("Workflow Results:")
 Results are items that result from a task():
           print(f"{task}: {str(result)[:150]}...")
          
 Excluded from e.
       print(f"❌ Error during execution: {str(e)}")
       print("💡 Tip: Make sure your Gemini API key is valid and you have internet connection")


def quick_scrape(url: str, instructions: str = "Extract main content"):
   """Quick scraping function for simple data extraction"""
   with AdvancedNotteAgent(headless=True, max_steps=5) as agent:
       response = agent.agent.run(
           task=f"{instructions} from this webpage",
           url=url
       )
 Answer back


def quick_search(query: str, num_results: int = 5):
   """Quick search function with structured results"""
   with AdvancedNotteAgent(headless=True, max_steps=10) as agent:
 Task = f"Search for '{query}' and return the top {num_results} results with titles, URLs, and brief descriptions."
       response = agent.agent.run(
           task=task,
           url="https://google.com",
           response_format=SearchResult
       )
 Answer back


def quick_form_fill(form_url: str, form_data: dict):
   """Quick form filling function"""
   with AdvancedNotteAgent(headless=False, max_steps=15) as agent:
       data_str = ", ".join([f"{k}: {v}" for k, v in form_data.items()])
 Task = F"Fill out the form with this information: {data_str}, then submit it."
      
       response = agent.agent.run(
           task=task,
           url=form_url
       )
 Answer back


If the __name__ equals "__main__":
   print("🧪 Quick Test Examples:")
   print("=" * 30)
  
   print("1. Quick Scrape Example:")
   try:
       result = quick_scrape("https://news.ycombinator.com", "Extract the top 3 post titles")
       print(f"Scraped: {result}")
 Except Exception As e.
       print(f"Error: {e}")
  
   print("n2. Quick Search Example:")
   try:
       search_results = quick_search("latest AI news", 3)
       print(f"Search Results: {search_results}")
 Exception to the rule:
       print(f"Error: {e}")
  
   print("n3. Custom Agent Task:")
   try:
       with AdvancedNotteAgent(headless=True) as agent:
           response = agent.agent.run(
               task="Go to Wikipedia, search for 'artificial intelligence', and summarize the main article in 2 sentences.",
               url="https://wikipedia.org"
           )
           print(f"Wikipedia Summary: {response.answer}")
 Except Exception As e.
       print(f"Error: {e}")
  
 You can also read more about it here.()
  
   print("n✨ Tutorial Complete!")
   print("💡 Tips for success:")
   print("- Start with simple tasks and gradually increase complexity")
   print("- Use structured outputs (Pydantic models) for reliable data extraction")
   print("- Implement rate limiting to respect API quotas")
   print("- Handle errors gracefully in production workflows")
   print("- Combine scripting with AI for cost-effective automation")
  
   print("n🚀 Next Steps:")
   print("- Customize the agents for your specific use cases")
   print("- Add error handling and retry logic for production")
   print("- Implement logging and monitoring for agent activities")
   print("- Scale up with Notte's hosted API service for enterprise features")