How to Build a Complete Multi-Domain AI Web Agent Using Notte and Gemini

!pip set up notte python-dotenv pydantic google-generativeai requests beautifulsoup4
!patchright set up --with-deps chromium


import os
import json
import time
from typing import List, Optional
from pydantic import BaseModel
import google.generativeai as genai
from 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):
   title: str
   worth: str
   ranking: Optional[float]
   availability: str
   description: str


class NewsArticle(BaseModel):
   title: str
   abstract: str
   url: str
   date: str
   supply: str


class SocialMediaPut up(BaseModel):
   content material: str
   creator: str
   likes: int
   timestamp: str
   platform: str


class SearchEnd result(BaseModel):
   question: str
   outcomes: 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, web site: str = "amazon.com") -> ProductInfo:
       """Research a product and extract structured data"""
       process = f"Go to {web site}, seek for '{product_name}', click on on the primary related product, and extract detailed product data together with title, worth, ranking, availability, and description."
      
       response = self.agent.run(
           process=process,
           response_format=ProductInfo,
           url=f"https://{web site}"
       )
       return response.reply


   def news_aggregator(self, subject: str, num_articles: int = 3) -> List[NewsArticle]:
       """Aggregate information articles on a particular subject"""
       process = f"Search for latest information about '{subject}', discover {num_articles} related articles, and extract title, abstract, URL, date, and supply for every."
      
       response = self.agent.run(
           process=process,
           url="https://information.google.com",
           response_format=List[NewsArticle]
       )
       return response.reply


   def social_media_monitor(self, hashtag: str, platform: str = "twitter") -> List[SocialMediaPost]:
       """Monitor social media for particular hashtags"""
       if platform.decrease() == "twitter":
           url = "https://twitter.com"
       elif platform.decrease() == "reddit":
           url = "https://reddit.com"
       else:
           url = f"https://{platform}.com"
          
       process = f"Go to {platform}, seek for posts with hashtag '{hashtag}', and extract content material, creator, engagement metrics, and timestamps from the highest 5 posts."
      
       response = self.agent.run(
           process=process,
           url=url,
           response_format=List[SocialMediaPost]
       )
       return response.reply


   def competitive_analysis(self, firm: str, opponents: List[str]) -> dict:
       """Perform aggressive evaluation by gathering pricing and characteristic information"""
       outcomes = {}
      
       for competitor in [company] + opponents:
           process = f"Go to {competitor}'s web site, discover their pricing web page or foremost product web page, and extract key options, pricing tiers, and distinctive promoting factors."
          
           attempt:
               response = self.agent.run(
                   process=process,
                   url=f"https://{competitor}.com"
               )
               outcomes[competitor] = response.reply
               time.sleep(2) 
           besides Exception as e:
               outcomes[competitor] = f"Error: {str(e)}"
              
       return outcomes


   def job_market_scanner(self, job_title: str, location: str = "distant") -> List[dict]:
       """Scan job marketplace for alternatives"""
       process = f"Search for '{job_title}' jobs in '{location}', extract job titles, firms, wage ranges, and utility URLs from the primary 10 outcomes."
      
       response = self.agent.run(
           process=process,
           url="https://certainly.com"
       )
       return response.reply


   def price_comparison(self, product: str, web sites: List[str]) -> dict:
       """Compare costs throughout a number of web sites"""
       price_data = {}
      
       for web site in web sites:
           process = f"Search for '{product}' on this web site and discover the most effective worth, together with any reductions or particular gives."
          
           attempt:
               response = self.agent.run(
                   process=process,
                   url=f"https://{web site}"
               )
               price_data[site] = response.reply
               time.sleep(1)
           besides Exception as e:
               price_data[site] = f"Error: {str(e)}"
              
       return price_data


   def content_research(self, subject: str, content_type: str = "weblog") -> dict:
       """Research content material concepts and trending subjects"""
       if content_type == "weblog":
           url = "https://medium.com"
           process = f"Search for '{subject}' articles, analyze trending content material, and establish fashionable themes, engagement patterns, and content material gaps."
       elif content_type == "video":
           url = "https://youtube.com"
           process = f"Search for '{subject}' movies, analyze view counts, titles, and descriptions to establish trending codecs and fashionable angles."
       else:
           url = "https://google.com"
           process = f"Search for '{subject}' content material throughout the net and analyze trending discussions and fashionable codecs."
          
       response = self.agent.run(process=process, url=url)
       return {"subject": subject, "insights": response.reply, "platform": content_type}

def demo_ecommerce_research():
   """Demo: E-commerce product analysis and comparability"""
   print(" E-commerce Research Demo")
   print("=" * 50)
  
   with AdvancedNotteAgent(headless=True) as agent:
       product = agent.research_product("wi-fi earbuds", "amazon.com")
       print(f"Product Research Results:")
       print(f"Name: {product.title}")
       print(f"Price: {product.worth}")
       print(f"Rating: {product.ranking}")
       print(f"Availability: {product.availability}")
       print(f"Description: {product.description[:100]}...")
      
       print("n Price Comparison:")
       web sites = ["amazon.com", "ebay.com", "walmart.com"]
       costs = agent.price_comparison("wi-fi earbuds", web sites)
       for web site, information in costs.objects():
           print(f"{web site}: {information}")


def demo_news_intelligence():
   """Demo: News aggregation and evaluation"""
   print(" News Intelligence Demo")
   print("=" * 50)
  
   with AdvancedNotteAgent() as agent:
       articles = agent.news_aggregator("synthetic intelligence", 3)
      
       for i, article in enumerate(articles, 1):
           print(f"nArticle {i}:")
           print(f"Title: {article.title}")
           print(f"Source: {article.supply}")
           print(f"Summary: {article.abstract}")
           print(f"URL: {article.url}")


def demo_social_listening():
   """Demo: Social media monitoring and sentiment evaluation"""
   print(" Social Media Listening Demo")
   print("=" * 50)
  
   with AdvancedNotteAgent() as agent:
       posts = agent.social_media_monitor("#AI", "reddit")
      
       for i, publish in enumerate(posts, 1):
           print(f"nPost {i}:")
           print(f"Author: {publish.creator}")
           print(f"Content: {publish.content material[:100]}...")
           print(f"Engagement: {publish.likes} likes")
           print(f"Platform: {publish.platform}")


def demo_market_intelligence():
   """Demo: Competitive evaluation and market analysis"""
   print(" Market Intelligence Demo")
   print("=" * 50)
  
   with AdvancedNotteAgent() as agent:
       firm = "openai"
       opponents = ["anthropic", "google"]
       evaluation = agent.competitive_analysis(firm, opponents)
      
       for comp, information in evaluation.objects():
           print(f"n{comp.higher()}:")
           print(f"Analysis: {str(information)[:200]}...")


def demo_job_market_analysis():
   """Demo: Job market scanning and evaluation"""
   print(" Job Market Analysis Demo")
   print("=" * 50)
  
   with AdvancedNotteAgent() as agent:
       jobs = agent.job_market_scanner("python developer", "san francisco")
      
       print(f"Found {len(jobs)} job alternatives:")
       for job in jobs[:3]:  
           print(f"- {job}")


def demo_content_strategy():
   """Demo: Content analysis and development evaluation"""
   print(" Content Strategy Demo")
   print("=" * 50)
  
   with AdvancedNotteAgent() as agent:
       blog_research = agent.content_research("machine studying", "weblog")
       video_research = agent.content_research("machine studying", "video")
      
       print("Blog Content Insights:")
       print(blog_research["insights"][:300] + "...")
      
       print("nVideo Content Insights:")
       print(video_research["insights"][:300] + "...")

class WorkflowSupervisor:
   def __init__(self):
       self.brokers = []
       self.outcomes = {}
      
   def add_agent_task(self, title: str, task_func, *args, **kwargs):
       """Add an agent process to the workflow"""
       self.brokers.append({
           'title': title,
           'func': task_func,
           'args': args,
           'kwargs': kwargs
       })
      
   def execute_workflow(self, parallel=False):
       """Execute all agent duties within the workflow"""
       print(" Executing Multi-Agent Workflow")
       print("=" * 50)
      
       for agent_task in self.brokers:
           title = agent_task['name']
           func = agent_task['func']
           args = agent_task['args']
           kwargs = agent_task['kwargs']
          
           print(f"n Executing {title}...")
           attempt:
               consequence = func(*args, **kwargs)
               self.outcomes[name] = consequence
               print(f" {title} accomplished efficiently")
           besides Exception as e:
               self.outcomes[name] = f"Error: {str(e)}"
               print(f" {title} failed: {str(e)}")
              
           if not parallel:
               time.sleep(2) 
       return self.outcomes


def market_research_workflow(company_name: str, product_category: str):
   """Complete market analysis workflow"""
   workflow = WorkflowSupervisor()
  
   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(class: str):
   """Research trending merchandise in a class"""
   with AdvancedNotteAgent(headless=True) as agent:
       process = f"Research trending {class} merchandise, discover high 5 merchandise with costs, scores, and key options."
       response = agent.agent.run(
           process=process,
           url="https://amazon.com"
       )
       return response.reply


def analyze_competitors(firm: str, class: str):
   """Analyze opponents available in the market"""
   with AdvancedNotteAgent(headless=True) as agent:
       process = f"Research {firm} opponents in {class}, evaluate pricing methods, options, and market positioning."
       response = agent.agent.run(
           process=process,
           url="https://google.com"
       )
       return response.reply


def monitor_brand_sentiment(model: str):
   """Monitor model sentiment throughout platforms"""
   with AdvancedNotteAgent(headless=True) as agent:
       process = f"Search for latest mentions of {model} on social media and information, analyze sentiment and key themes."
       response = agent.agent.run(
           process=process,
           url="https://reddit.com"
       )
       return response.reply

def foremost():
   """Main perform to run all demos"""
   print(" Advanced Notte AI Agent Tutorial")
   print("=" * 60)
   print("Note: Make positive 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 within the code above!")
       return
  
   attempt:
       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")
       outcomes = market_research_workflow("Tesla", "electrical automobiles")
       print("Workflow Results:")
       for process, end in outcomes.objects():
           print(f"{process}: {str(consequence)[:150]}...")
          
   besides Exception as e:
       print(f" Error throughout execution: {str(e)}")
       print(" Tip: Make positive your Gemini API secret's legitimate and you may have web connection")


def quick_scrape(url: str, directions: str = "Extract foremost content material"):
   """Quick scraping perform for easy information extraction"""
   with AdvancedNotteAgent(headless=True, max_steps=5) as agent:
       response = agent.agent.run(
           process=f"{directions} from this webpage",
           url=url
       )
       return response.reply


def quick_search(question: str, num_results: int = 5):
   """Quick search perform with structured outcomes"""
   with AdvancedNotteAgent(headless=True, max_steps=10) as agent:
       process = f"Search for '{question}' and return the highest {num_results} outcomes with titles, URLs, and temporary descriptions."
       response = agent.agent.run(
           process=process,
           url="https://google.com",
           response_format=SearchEnd result
       )
       return response.reply


def quick_form_fill(form_url: str, form_data: dict):
   """Quick kind filling perform"""
   with AdvancedNotteAgent(headless=False, max_steps=15) as agent:
       data_str = ", ".be part of([f"{k}: {v}" for k, v in form_data.items()])
       process = f"Fill out the shape with this data: {data_str}, then submit it."
      
       response = agent.agent.run(
           process=process,
           url=form_url
       )
       return response.reply


if __name__ == "__main__":
   print(" Quick Test Examples:")
   print("=" * 30)
  
   print("1. Quick Scrape Example:")
   attempt:
       consequence = quick_scrape("https://information.ycombinator.com", "Extract the highest 3 publish titles")
       print(f"Scraped: {consequence}")
   besides Exception as e:
       print(f"Error: {e}")
  
   print("n2. Quick Search Example:")
   attempt:
       search_results = quick_search("newest AI information", 3)
       print(f"Search Results: {search_results}")
   besides Exception as e:
       print(f"Error: {e}")
  
   print("n3. Custom Agent Task:")
   attempt:
       with AdvancedNotteAgent(headless=True) as agent:
           response = agent.agent.run(
               process="Go to Wikipedia, seek for 'synthetic intelligence', and summarize the principle article in 2 sentences.",
               url="https://wikipedia.org"
           )
           print(f"Wikipedia Summary: {response.reply}")
   besides Exception as e:
       print(f"Error: {e}")
  
   foremost()
  
   print("n Tutorial Complete!")
   print(" Tips for achievement:")
   print("- Start with easy duties and progressively enhance complexity")
   print("- Use structured outputs (Pydantic fashions) for dependable information extraction")
   print("- Implement fee limiting to respect API quotas")
   print("- Handle errors gracefully in manufacturing workflows")
   print("- Combine scripting with AI for cost-effective automation")
  
   print("n Next Steps:")
   print("- Customize the brokers in your particular use circumstances")
   print("- Add error dealing with and retry logic for manufacturing")
   print("- Implement logging and monitoring for agent actions")
   print("- Scale up with Notte's hosted API service for enterprise options")