┌─────────────────────────────────────────────────┐
│                                                 │
│  ┌─────────────┐        ┌─────────────────┐    │
│  │ news-hunter │        │   commentator   │    │
│  │   (Python)  │        │     (YAML)      │    │
│  │             │        │                 │    │
│  │ Hacker News ├───────►│ Analysis &      │    │
│  │ API fetch   │  post  │ Hot Takes       │    │
│  └─────────────┘        └─────────────────┘    │
│         │                       │              │
│         └───────────┬───────────┘              │
│                     ▼                          │
│           news-feed channel                    │
└─────────────────────────────────────────────────┘

# Optional: Set the OpenAI base URL
export OPENAI_BASE_URL="your-base-url-here"
 
# Must: Set the OpenAI API key
export OPENAI_API_KEY="your-key-here"

openagents network start demos/02_tech_news_stream/

python demos/02_tech_news_stream/agents/news_hunter.py

python demos/02_tech_news_stream/agents/news_hunter.py --interval 120 --host localhost --port 8700

openagents agent start demos/02_tech_news_stream/agents/commentator.yaml

# demos/02_tech_news_stream/agents/news_hunter.py
from openagents.agents.worker_agent import WorkerAgent
from tools.news_fetcher import fetch_hackernews_top
 
class NewsHunterAgent(WorkerAgent):
    """A news hunter that continuously fetches and posts tech news."""
 
    default_agent_id = "news-hunter"
 
    def __init__(self, fetch_interval: int = 60, **kwargs):
        super().__init__(**kwargs)
        self.fetch_interval = fetch_interval
        self.posted_urls = set()  # Track posted URLs to avoid duplicates
        self._hunting_task = None
 
    async def on_startup(self):
        """Called when agent connects to the network."""
        print(f"News Hunter connected! Starting hunt loop...")
        self._hunting_task = asyncio.create_task(self._hunt_news_loop())
 
    async def on_shutdown(self):
        """Called when agent shuts down."""
        if self._hunting_task:
            self._hunting_task.cancel()
 
    async def _hunt_news_loop(self):
        """Continuous loop to fetch and post news."""
        await asyncio.sleep(5)  # Wait for initialization
 
        while True:
            await self._fetch_and_post_news()
            await asyncio.sleep(self.fetch_interval)
 
    async def _fetch_and_post_news(self):
        """Fetch news and post new stories."""
        news_data = fetch_hackernews_top(count=5)
        stories = self._parse_news(news_data)
 
        # Only post stories we haven't posted before
        new_stories = [s for s in stories if s['url'] not in self.posted_urls]
 
        for story in new_stories[:2]:  # Post max 2 per cycle
            await self._post_story(story)
            self.posted_urls.add(story['url'])
 
    async def _post_story(self, story: dict):
        """Post a story to the news-feed channel."""
        message = f"**{story['title']}**\n\n{story['url']}\n{story['score']} points"
 
        messaging = self.client.mod_adapters.get("openagents.mods.workspace.messaging")
        if messaging:
            await messaging.send_channel_message(
                channel="news-feed",
                text=message
            )

# demos/02_tech_news_stream/tools/news_fetcher.py
 
def fetch_hackernews_top(count: int = 5) -> str:
    """Fetch top stories from Hacker News."""
    response = requests.get(
        "https://hacker-news.firebaseio.com/v0/topstories.json"
    )
    story_ids = response.json()[:count]
    # ... fetch and format each story
    return formatted_stories
 
def fetch_hackernews_new(count: int = 5) -> str:
    """Fetch newest stories from Hacker News."""
    # Similar implementation for new stories
 
def fetch_url_content(url: str, max_length: int = 5000) -> str:
    """Fetch and extract text content from a URL."""
    # HTML extraction and cleaning

# demos/02_tech_news_stream/agents/commentator.yaml
type: "openagents.agents.collaborator_agent.CollaboratorAgent"
agent_id: "commentator"
 
config:
  model_name: "gpt-4o-mini"
 
  instruction: |
    You are the COMMENTATOR - a sharp tech analyst.
 
    COMMENTARY STYLES:
    - For AI/ML news: Discuss implications, compare to breakthroughs
    - For Startup news: Analyze market opportunity, predict challenges
    - For Big Tech news: Consider antitrust angle, ecosystem impact
    - For Security news: Practical advice, risk assessment
 
    RESPONSE FORMAT:
    - Use emojis for hot takes, analysis, predictions, concerns
 
  # Don't react to all messages - use triggers instead
  react_to_all_messages: false
 
  triggers:
    - event: "thread.channel_message.notification"
      instruction: |
        A new message in news-feed. Read and reply with commentary.
 
mods:
  - name: "openagents.mods.workspace.messaging"
    enabled: true

def fetch_reddit_tech(subreddit: str = "technology", count: int = 5) -> str:
    """Fetch from Reddit tech subreddits."""
    # Implementation
 
def fetch_techcrunch(count: int = 5) -> str:
    """Fetch from TechCrunch RSS."""
    # Implementation

agent_id: "summarizer"
config:
  instruction: |
    At the end of each hour, summarize the key news stories posted.
    Identify trends and common themes across stories.