How to Build a Trend Forecasting Tool with Social Scraping

Trends emerge on social media before hitting mainstream. By scraping platforms systematically, you can detect rising trends days before they become obvious.

Data Collection

import requests
from bs4 import BeautifulSoup
from datetime import datetime, timedelta
import sqlite3, time

class TrendCollector:
    def __init__(self, db_path='trends.db', api_key=None):
        self.db = sqlite3.connect(db_path)
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({'User-Agent': 'TrendResearch/1.0'})
        self._init_db()

    def _init_db(self):
        self.db.executescript('''
            CREATE TABLE IF NOT EXISTS mentions (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                platform TEXT, keyword TEXT, content TEXT,
                engagement INTEGER, timestamp DATETIME, url TEXT);
            CREATE INDEX IF NOT EXISTS idx_kw ON mentions(keyword, timestamp);
        ''')

    def _fetch(self, url):
        if self.api_key:
            return self.session.get(
                f"http://api.scraperapi.com?api_key={self.api_key}&url={url}&render=true")
        return self.session.get(url)

    def collect_reddit(self, keyword, subs=None):
        subs = subs or ['all']
        mentions = []
        for sub in subs:
            resp = self._fetch(
                f"https://old.reddit.com/r/{sub}/search?q={keyword}&sort=new&t=week")
            soup = BeautifulSoup(resp.text, 'html.parser')
            for post in soup.select('.thing.link'):
                title = post.select_one('a.title')
                score = post.select_one('.score.unvoted')
                if title:
                    eng = 0
                    if score:
                        try: eng = int(score.get_text(strip=True))
                        except: pass
                    mentions.append({
                        'platform': 'reddit', 'keyword': keyword,
                        'content': title.get_text(strip=True),
                        'engagement': eng, 'url': title.get('href',''),
                        'timestamp': datetime.now().isoformat()
                    })
            time.sleep(2)
        for m in mentions:
            self.db.execute(
                'INSERT INTO mentions (platform,keyword,content,engagement,timestamp,url) VALUES (?,?,?,?,?,?)',
                (m['platform'],m['keyword'],m['content'],m['engagement'],m['timestamp'],m['url']))
        self.db.commit()
        return mentions

Velocity Engine

The key insight: acceleration matters more than volume.

class VelocityEngine:
    def __init__(self, db):
        self.db = db

    def velocity(self, keyword, hours=24):
        now = datetime.now()
        cur_start = now - timedelta(hours=hours)
        prev_start = cur_start - timedelta(hours=hours)
        cur = self._count(keyword, cur_start, now)
        prev = self._count(keyword, prev_start, cur_start)
        if prev == 0: return float('inf') if cur > 0 else 0
        return round((cur - prev) / prev, 4)

    def _count(self, kw, start, end):
        c = self.db.execute(
            'SELECT COUNT(*) FROM mentions WHERE keyword=? AND timestamp BETWEEN ? AND ?',
            (kw, start.isoformat(), end.isoformat()))
        return c.fetchone()[0]

    def engagement_velocity(self, keyword, hours=24):
        now = datetime.now()
        cur_start = now - timedelta(hours=hours)
        prev_start = cur_start - timedelta(hours=hours)
        cur = self._engagement(keyword, cur_start, now)
        prev = self._engagement(keyword, prev_start, cur_start)
        if prev == 0: return float('inf') if cur > 0 else 0
        return round((cur - prev) / prev, 4)

    def _engagement(self, kw, start, end):
        c = self.db.execute(
            'SELECT COALESCE(SUM(engagement),0) FROM mentions WHERE keyword=? AND timestamp BETWEEN ? AND ?',
            (kw, start.isoformat(), end.isoformat()))
        return c.fetchone()[0]

Breakout Detection

class BreakoutDetector:
    def __init__(self, engine):
        self.engine = engine

    def detect(self, keywords, threshold=0.5):
        results = []
        for kw in keywords:
            mv = self.engine.velocity(kw)
            ev = self.engine.engagement_velocity(kw)
            score = (mv * 0.4) + (ev * 0.6)
            if score > threshold:
                results.append({'keyword': kw, 'mention_vel': mv,
                               'engagement_vel': ev, 'score': round(score, 3)})
        return sorted(results, key=lambda x: x['score'], reverse=True)

Trend Forecasting

class Forecaster:
    def predict(self, keyword, db, points=7):
        cursor = db.execute('''
            SELECT DATE(timestamp), COUNT(*) FROM mentions
            WHERE keyword=? GROUP BY DATE(timestamp)
            ORDER BY DATE(timestamp) DESC LIMIT ?
        ''', (keyword, points))
        data = cursor.fetchall()
        if len(data) < 3: return {'prediction': 'insufficient_data'}
        vols = [r[1] for r in reversed(data)]
        n = len(vols)
        xm = (n-1)/2
        ym = sum(vols)/n
        slope = sum((i-xm)*(v-ym) for i,v in enumerate(vols))
        slope /= sum((i-xm)**2 for i in range(n))
        rate = slope/ym if ym > 0 else 0
        if rate > 0.15: p = 'accelerating'
        elif rate > 0.05: p = 'growing'
        elif rate > -0.05: p = 'stable'
        else: p = 'declining'
        return {'keyword': keyword, 'prediction': p, 'growth_rate': round(rate, 4)}

Social platforms are hard to scrape. ScraperAPI handles anti-bot protections. ThorData provides residential rotation. Track rates with ScrapeOps.

Follow for more Python data science tutorials.

How to Scrape Wayback Machine: Historical Web Data with Python

The Wayback Machine stores over 800 billion web pages dating back to 1996. This data is invaluable for research, competitive analysis, content recovery, and tracking website evolution.

CDX API: The Power Tool

The Wayback Machine provides a CDX API returning structured data about archived URLs — no scraping needed for the index.

import requests, json, time
from datetime import datetime
from bs4 import BeautifulSoup
import difflib

class WaybackScraper:
    CDX_API = "http://web.archive.org/cdx/search/cdx"
    WEB_BASE = "http://web.archive.org/web"

    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({'User-Agent': 'WaybackResearch/1.0'})

    def get_snapshots(self, url, from_date=None, to_date=None, limit=1000):
        params = {'url': url, 'output': 'json', 'limit': limit,
                  'fl': 'timestamp,original,statuscode,mimetype,length'}
        if from_date: params['from'] = from_date
        if to_date: params['to'] = to_date
        resp = self.session.get(self.CDX_API, params=params)
        data = resp.json()
        if len(data) <= 1: return []
        headers = data[0]
        return [dict(zip(headers, row)) for row in data[1:]]

    def get_page(self, url, timestamp):
        return self.session.get(f"{self.WEB_BASE}/{timestamp}id_/{url}").text

Tracking Changes Over Time

class EvolutionTracker:
    def __init__(self, scraper):
        self.scraper = scraper

    def track_changes(self, url, selector, samples=10):
        snaps = self.scraper.get_snapshots(url)
        if not snaps: return []
        step = max(1, len(snaps) // samples)
        versions = []
        for snap in snaps[::step]:
            try:
                html = self.scraper.get_page(url, snap['timestamp'])
                soup = BeautifulSoup(html, 'html.parser')
                el = soup.select_one(selector)
                if el:
                    versions.append({
                        'timestamp': snap['timestamp'],
                        'date': datetime.strptime(snap['timestamp'][:8], '%Y%m%d').isoformat(),
                        'content': el.get_text(strip=True)[:500]
                    })
                time.sleep(1)
            except Exception as e:
                print(f"Error: {e}")
        return versions

    def compare(self, versions):
        changes = []
        for i in range(1, len(versions)):
            old = versions[i-1]['content'].splitlines()
            new = versions[i]['content'].splitlines()
            diff = list(difflib.unified_diff(old, new, lineterm=''))
            if diff:
                changes.append({'from': versions[i-1]['date'], 'to': versions[i]['date'],
                                'diff_lines': len(diff)})
        return changes

Bulk Collection

class BulkCollector:
    def __init__(self, scraper):
        self.scraper = scraper

    def domain_history(self, domain):
        params = {'url': domain, 'output': 'json', 'matchType': 'domain',
                  'fl': 'timestamp,original,statuscode', 'collapse': 'urlkey',
                  'filter': 'statuscode:200'}
        data = self.scraper.session.get(self.scraper.CDX_API, params=params).json()
        if len(data) <= 1: return []
        return [dict(zip(data[0], row)) for row in data[1:]]

    def download_snapshot(self, domain, timestamp, output_dir='wayback_data'):
        import os
        os.makedirs(output_dir, exist_ok=True)
        urls = self.domain_history(domain)
        count = 0
        for entry in urls[:100]:
            try:
                content = self.scraper.get_page(entry['original'], timestamp)
                path = entry['original'].replace(f'https://{domain}','').replace(f'http://{domain}','')
                path = path.strip('/') or 'index.html'
                fp = os.path.join(output_dir, path)
                os.makedirs(os.path.dirname(fp), exist_ok=True)
                with open(fp, 'w') as f: f.write(content)
                count += 1
                time.sleep(1)
            except: pass
        return count

Competitor Pricing History

def track_pricing(scraper, url, selector):
    tracker = EvolutionTracker(scraper)
    versions = tracker.track_changes(url, selector, samples=20)
    print(f"Found {len(versions)} snapshots")
    for v in versions:
        print(f"  {v['date']}: {v['content'][:100]}")
    changes = tracker.compare(versions)
    print(f"Detected {len(changes)} changes")
    return versions, changes

wb = WaybackScraper()
track_pricing(wb, 'https://example.com/pricing', '.pricing-table')

Tips

1. Use CDX API first — fast, no page loads needed
2. Add 1-2s delays — Internet Archive is a nonprofit
3. Use collapse to deduplicate
4. Use id_ modifier for original pages without toolbar
5. Cache results locally

For large-scale collection, ScraperAPI handles proxy rotation. ThorData provides IP diversity. Monitor with ScrapeOps.

Follow for more Python data collection tutorials.

Building a Competitive Intelligence Dashboard with Web Scraping

Understanding competitors’ pricing changes, hiring patterns, and marketing strategies gives you a strategic advantage. Here’s how to build a Python-powered competitive intelligence system.

Core Engine

import requests
from bs4 import BeautifulSoup
import sqlite3, json, hashlib
from datetime import datetime
import time, re

class CompetitiveIntel:
    def __init__(self, db_path='intel.db', api_key=None):
        self.db = sqlite3.connect(db_path)
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'})
        self._init_db()

    def _init_db(self):
        self.db.executescript('''
            CREATE TABLE IF NOT EXISTS competitors (
                id TEXT PRIMARY KEY, name TEXT, domain TEXT, category TEXT);
            CREATE TABLE IF NOT EXISTS intel (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                competitor_id TEXT, dimension TEXT,
                title TEXT, content TEXT, url TEXT,
                content_hash TEXT, detected_at DATETIME,
                UNIQUE(competitor_id, content_hash));
            CREATE TABLE IF NOT EXISTS pricing (
                competitor_id TEXT, plan TEXT,
                price REAL, features TEXT, captured_at DATETIME);
        ''')

    def _fetch(self, url):
        if self.api_key:
            return self.session.get(
                f"http://api.scraperapi.com?api_key={self.api_key}&url={url}&render=true")
        return self.session.get(url)

    def add_competitor(self, cid, name, domain, cat='direct'):
        self.db.execute('INSERT OR REPLACE INTO competitors VALUES (?,?,?,?)',
                       (cid, name, domain, cat))
        self.db.commit()

Pricing Intelligence

    def track_pricing(self, cid, url, plan_sel, price_sel):
        resp = self._fetch(url)
        soup = BeautifulSoup(resp.text, 'html.parser')
        plans = []
        for el in soup.select(plan_sel):
            name = el.select_one('h2, h3, .plan-name')
            price_el = el.select_one(price_sel)
            if not name or not price_el: continue
            match = re.search(r'$(d+.?d*)', price_el.get_text())
            price = float(match.group(1)) if match else 0
            features = [li.get_text(strip=True) for li in el.select('li')]
            plans.append({'name': name.get_text(strip=True), 'price': price, 'features': features})
            self.db.execute('INSERT INTO pricing VALUES (?,?,?,?,?)',
                (cid, plans[-1]['name'], price, json.dumps(features), datetime.now().isoformat()))
        self.db.commit()
        return plans

    def pricing_changes(self, cid):
        rows = self.db.execute(
            'SELECT plan, price, captured_at FROM pricing WHERE competitor_id=? ORDER BY captured_at DESC',
            (cid,)).fetchall()
        changes, seen = [], {}
        for plan, price, date in rows:
            if plan in seen and seen[plan] != price:
                changes.append({'plan': plan, 'old': price, 'new': seen[plan],
                    'pct': round((seen[plan]-price)/price*100, 1)})
            seen[plan] = price
        return changes

Job Posting Intelligence

    def track_hiring(self, cid, url, job_sel):
        resp = self._fetch(url)
        soup = BeautifulSoup(resp.text, 'html.parser')
        jobs = []
        for el in soup.select(job_sel):
            title = el.select_one('h2, h3, a, .job-title')
            dept = el.select_one('.department, .team')
            loc = el.select_one('.location')
            if not title: continue
            job = {'title': title.get_text(strip=True),
                   'dept': dept.get_text(strip=True) if dept else 'Unknown',
                   'location': loc.get_text(strip=True) if loc else 'Unknown'}
            jobs.append(job)
            h = hashlib.sha256(job['title'].encode()).hexdigest()
            try:
                self.db.execute('INSERT INTO intel VALUES (NULL,?,?,?,?,?,?,?)',
                    (cid, 'hiring', job['title'], json.dumps(job), '', h, datetime.now().isoformat()))
            except sqlite3.IntegrityError: pass
        self.db.commit()
        return jobs

    def hiring_trends(self, cid):
        rows = self.db.execute('''
            SELECT content FROM intel WHERE competitor_id=? AND dimension='hiring'
            AND detected_at > datetime('now','-30 days')
        ''', (cid,)).fetchall()
        depts = {}
        for r in rows:
            d = json.loads(r[0]).get('dept', 'Unknown')
            depts[d] = depts.get(d, 0) + 1
        return dict(sorted(depts.items(), key=lambda x: x[1], reverse=True))

Reports

class IntelReport:
    def __init__(self, engine):
        self.e = engine

    def generate(self, cid):
        comp = self.e.db.execute('SELECT name,domain FROM competitors WHERE id=?', (cid,)).fetchone()
        if not comp: return None
        return {'competitor': comp[0], 'domain': comp[1],
                'pricing_changes': self.e.pricing_changes(cid),
                'hiring_trends': self.e.hiring_trends(cid)}

    def display(self, r):
        print(f"nIntelligence: {r['competitor']}")
        print("="*50)
        if r['pricing_changes']:
            print("nPricing Changes:")
            for c in r['pricing_changes']:
                d = 'UP' if c['pct'] > 0 else 'DOWN'
                print(f"  {c['plan']}: ${c['old']} -> ${c['new']} ({d} {abs(c['pct'])}%)")
        if r['hiring_trends']:
            print("nHiring:")
            for dept, n in r['hiring_trends'].items():
                print(f"  {dept}: {n} positions")

intel = CompetitiveIntel(api_key='YOUR_KEY')
intel.add_competitor('c1', 'Competitor A', 'competitor-a.com')
report = IntelReport(intel)
r = report.generate('c1')
report.display(r)

For monitoring dozens of competitors, ScraperAPI handles JS rendering. ThorData provides residential proxies. Track pipelines with ScrapeOps.

Follow for more Python business intelligence tutorials.

In 2026, access to powerful AI assistants is widespread. That can feel like the playing field has leveled—until you notice the same pattern in hiring and promotions: outputs look similar, but outcomes don’t.

The gap is rarely “who has the tool.” It’s who can steer it with context, standards, and accountability.

Same shortcuts, different standards

When baseline drafting, summarizing, and brainstorming become cheap, what separates strong contributors is not speed alone. It’s the ability to:

• Define the problem clearly enough that automation doesn’t optimize the wrong thing
• Judge quality against domain constraints (legal, medical, financial, brand, security)
• Iterate with intent instead of accepting the first plausible answer
• Communicate trade-offs to stakeholders who don’t care about models—only results

Tools amplify habits. If your habits are shallow, you get shallow work—just faster.

“Knowing AI” is mostly knowing your job

The durable skill set looks less like memorized prompts and more like professional judgment:

• What must be verified, and what counts as a trustworthy source?
• Where does automation create risk (privacy, compliance, reputation)?
• How do you document decisions so a team can audit and improve workflows?

This is why organizations increasingly reward people who can combine domain expertise with structured experimentation—not people who treat AI like a magic button.

Learning that matches how work actually happens

Self-guided tinkering is fine for curiosity, but if you want progress you can measure, it helps to follow a path that connects concepts to repeatable practice. For learners who want that structure in Romanian, Cursuri AI is built around applied progression rather than scattered tips.

If you’re the kind of reader who prefers to understand the flow before diving in, the walkthrough on how it works is a straightforward way to see how the experience is organized end to end.

A practical weekly rhythm (that compounds)

If you want a lightweight habit that actually builds skill:

1. Pick one recurring task you do every week
2. Improve it twice: once for speed, once for quality (checklist + review)
3. Keep a short log of failures—hallucinations, wrong assumptions, missed edge cases
4. Once a month, refactor your workflow based on that log

Over a quarter, you’ll feel less like you’re “using AI” and more like you’re running a repeatable process—which is what teams pay for.

Bottom line

When tools are ubiquitous, differentiation returns to taste, rigor, and responsibility. The winners aren’t the loudest adopters—they’re the ones who raise the quality bar while everyone else settles for “good enough at a glance.”

Part 1: The CI/CD Bottleneck Nobody Talks About

Platform Engineering matured how we provision infrastructure. It hasn’t touched how we build and deploy code.

CI/CD was, a decade ago, at the heart of the DevOps revolution. It deeply transformed the way we ship software by providing continuous automation workflows from code to production. It hasn’t changed much since then. And like every tool that has been continuously used for so long, the more years pass, the more the flaws and frustrations seem obvious. The “traditional” way of building CI/CD nowadays often feels cumbersome and rusty.

Among the almost always recurrent flaws that any DevOps engineer has witnessed:

• Lack of development ergonomics. For most technologies, the only way to really test pipelines is by pushing the code to the server, waiting, and iterating. It’s inefficient and time-consuming. The feedback loop for CI/CD code is worse than for any other code in your organization.

• YAML sprawl. As soon as pipelines require more logic and complexity, you rapidly end up with thousands of YAML files to maintain across repositories, each slightly different, each owned by whoever touched it last.

• Opaque failure investigation. When things go wrong in a pipeline, there is no intelligence. Developers sift through thousands of log lines to manually investigate, with no programmatic way to diagnose or recover.

Meanwhile, Infrastructure-as-Code solved provisioning. GitOps solved configuration drift. Internal Developer Platforms gave teams golden paths for spinning up services, databases, and environments. But open any repository in your organization and look at the CI/CD: it’s still bespoke YAML. In a world where platform teams build abstractions for everything else, CI/CD remains the one piece of developer infrastructure that every team reinvents from scratch.

This series explores what a CI/CD pipeline could look like in 2026, one that addresses all three of these flaws, at the intersection of Platform Engineering and AI agents. To do that, we’ll use one of the most exciting tools I’ve seen in the CI/CD landscape in a long time: Dagger. Dagger reimagines pipelines as typed functions written in real programming languages, running inside containers with deterministic caching, portable across any CI engine. But what makes it particularly relevant for this series is that it doesn’t stop at build automation: it has native LLM integration, a composable module system, and an emerging agent ecosystem that lets you go from reusable pipeline modules to AI-powered CI/CD workflows. It’s the right tool to explore what happens when Platform Engineering meets AI agents in the CI/CD space.

We’ll build pipelines as real code, decouple them from the infrastructure they run on, package them as reusable modules, and eventually compose AI agents that handle review, build, and deployment from a single natural language command.

The Series at a Glance

This is a 4-part series. Each part builds on the previous one, progressively transforming how we think about CI/CD:

1. Part 1 (this article): Pipelines as real code. We replace YAML with typed, testable, locally-runnable pipeline functions using Dagger, and build a complete CI/CD pipeline for an Angular + FastAPI stack deployed to Google Cloud.

2. Part 2: Decoupling Pipelines from Infrastructure: Same pipeline code, different runners. We explore three options for scaling: GitHub Actions (free tier), managed runners with Depot.dev, and self-hosted Kubernetes runners with ARC, including a Terraform module for production deployments.

3. Part 3: From Scripts to a Platform: Your CI/CD Module Library: Modules as the unit of CI/CD reuse. We build a two-layer architecture: public daggerverse modules for generic operations (GCP Auth, Cloud Run, Firebase) wrapped by private organizational modules that enforce compliance, security, and conventions.

4. Part 4: The AI-Native CI/CD Stack: Agents, Modules, and Spec-Driven Development: We leverage the agentic capabilities of Dagger to enhance the developer experience, automate CI failure remediation, and explore what an Internal Developer Factory looks like when built on Spec-Driven Development principles. The pipeline itself stays fast and deterministic, with no LLM in the hot path.

But first: let’s look at the root cause.

The YAML Illusion

YAML-based CI/CD looks simple. A few lines, a run: step, and you have a pipeline. But that simplicity is an illusion. The moment a real-world project needs conditional logic, secret management, reusable workflows, matrix builds, or cross-repository dependencies, every CI platform reaches for its own layer of abstractions on top of YAML.

GitHub Actions has reusable workflows, composite actions, and a marketplace of third-party actions you import by reference. GitLab CI has include:, extends:, and multi-project pipelines. CircleCI has orbs. Azure DevOps has templates and task groups. Each platform reinvented inheritance, composition, and parameterization (poorly) on top of a configuration format that was never designed for it.

The result is a system that is declarative in appearance but imperative in practice. A developer looking at a .github/workflows/deploy.yml that imports three reusable workflows, each with their own if: conditions and secret references, is not reading a simple configuration file. They’re reading a distributed program written in a language with no type system, no IDE support, no debugger, and no way to run it locally.

This is the fundamental problem: CI/CD, as it exists today, is not part of the software factory. It sits outside the development workflow. It can’t be run on a developer’s machine. It can’t be unit tested. It can’t be refactored with confidence. It can’t be debugged step by step. It doesn’t benefit from the tooling, the practices, or the discipline that every other piece of software in the organization does.

And yet it’s the most critical automation in your delivery chain, the one that decides whether code reaches production.

CI/CD as Software

The fix isn’t a better YAML syntax. It’s a paradigm shift: treat CI/CD as an integral part of the software factory.

That means CI/CD primitives (building a container, running tests, publishing an artifact, deploying a service) should be programmatic abstractions. Written in real programming languages. With types, functions, imports, error handling. Runnable on a developer’s laptop. Testable. Debuggable. Versioned alongside the application code. Reviewable in the same pull request.

This is what Dagger does. It decomposes CI/CD operations into typed functions, available through SDKs in Python, Go, and TypeScript. A pipeline is not a configuration file interpreted by a remote server. It’s code that runs inside containers, with content-addressed caching at every step.

A note on language choice — Throughout this series, we’ll use Python for all our examples. But everything we write here works identically in Go, TypeScript, or Java. Dagger’s SDKs expose the same API surface and the same types across all supported languages — the pipeline behavior is determined by the engine, not the SDK. Pick the language your team is most comfortable with.

from typing import Annotated

import dagger
from dagger import Doc, dag, function, object_type


@object_type
class MyPipeline:
    @function
    async def build(
        self,
        source: Annotated[dagger.Directory, Doc("Application source code")],
    ) -> dagger.Container:
        """Build the application container."""
        return (
            dag.container()
            .from_("python:3.12-slim")
            .with_directory("/app", source)
            .with_workdir("/app")
            .with_exec(["pip", "install", "-r", "requirements.txt"])
            .with_entrypoint(["python", "main.py"])
        )

This is a typed, composable function that:

• Runs on your laptop with dagger call build --source=.
• Runs in GitHub Actions with the exact same command
• Runs in GitLab CI with the exact same command
• Can be debugged, unit tested, and refactored like any other code

No YAML translation. No platform-specific quirks. The same code, the same command, everywhere.

The shift is subtle but significant. CI/CD is no longer a separate discipline with its own language, its own debugging workflow, and its own specialists. It’s software. Developers can read it, understand it, modify it, and run it, the same way they do with application code. That’s the self-service promise of Platform Engineering applied to the one area it hadn’t reached yet.

Three Properties That Make It Work

When CI/CD becomes software, three properties emerge that YAML pipelines fundamentally cannot provide.

1. Container-Based Execution

Every operation happens inside a container. When you write:

dag.container().from_("python:3.12-slim").with_exec(["pip", "install", "pytest"])

You’re building a container execution graph. The engine takes this graph and executes it efficiently. Your pipeline is reproducible (same inputs, same outputs), isolated (no “works on my machine”), and portable (containers run anywhere).

This is what enables local execution. The same container graph that runs on a CI runner runs on your laptop. No emulation, no mocking, just the real thing. A developer can iterate on a pipeline the same way they iterate on application code: change, run, observe, repeat. The feedback loop goes from “push and wait 10 minutes” to “run and see in seconds.”

2. Content-Addressed Caching

Every operation is cached based on the hash of its inputs:

@function
async def test(
    self,
    source: Annotated[dagger.Directory, Doc("Application source code")],
) -> str:
    """Run tests with intelligent caching."""
    return await (
        dag.container()
        .from_("python:3.12-slim")
        .with_directory("/app", source)       # Cached if source unchanged
        .with_exec(["pip", "install", "-r", "requirements.txt"])  # Cached if requirements unchanged
        .with_exec(["pytest"])                 # Only re-runs if dependencies changed
        .stdout()
    )

Change a test file? Only the test execution re-runs. Change requirements.txt? Dependencies reinstall, but the base image stays cached. This is content-addressed: Dagger hashes the actual content, not timestamps.

Because it’s deterministic, caching works identically on your laptop and in CI. A developer running dagger call test --source=. after changing one test file gets the same instant feedback a CI runner would, with no cold start and no full rebuild.

3. A Module System

Modules are reusable pipeline components that can be published and shared:

# Use a published module
dagger call -m github.com/purpleclay/daggerverse/[email protected] build --source=.

# Or reference your own
dagger call -m ./.dagger build --source=.

Modules have typed interfaces, documentation, and versioning. They compose:

@function
async def deploy(
    self,
    source: Annotated[dagger.Directory, Doc("Application source")],
) -> str:
    # Authenticate to GCP
    gcp = dag.gcp_auth().from_service_account_key(key=sa_key, project=project)

    # Publish to Artifact Registry
    image_uri = await dag.gcp_artifact_registry(gcp).publish(
        container=await self.build(source),
        repository="my-repo", image="my-app", tag="latest",
    )

    # Deploy to Cloud Run
    return await dag.gcp_cloud_run(gcp).service().deploy(
        name="my-service", image=image_uri, region="us-central1",
    )

This is the Platform Engineering pattern applied to CI/CD. A platform team builds and maintains modules (gcp-auth, artifact-registry, cloud-run) the same way they’d maintain Terraform modules or Helm charts. Developers consume them as typed dependencies. They don’t need to know how GCP authentication works internally; they call dag.gcp_auth() and get an authenticated container. Self-service, with guardrails.

Putting It Into Practice

Enough theory. For the rest of this series, we’re going to simulate, as closely as we can, a production deployment. Not a toy example with a single endpoint and a hello world container, but a realistic two-component product with a frontend, a backend, authentication between them, and deployment to a real cloud provider.

The goal is a deep dive: from writing the first pipeline function on your laptop, all the way to a multi-agent CI/CD system that reviews, builds, and deploys both apps from a single natural language command.

Prerequisites

To follow along, you’ll need three things set up:

1. Dagger CLI: Install the Dagger engine and CLI on your machine. This is what runs your pipelines locally and in CI.
   → Install Dagger

2. A GitHub account: We’ll use GitHub as our code server and CI runner (via GitHub Actions). The companion repository with all example code is public.
   → Create a GitHub account if you don’t have one

3. A Google Cloud account with a project: GCP is our deployment target throughout the series. You’ll need a project with billing enabled (the free tier is sufficient for everything we do here).
   → Create a GCP account
   → Create a GCP project

Required Google APIs — Enable these APIs on your GCP project before deploying. You can do it from the API Library or via gcloud:

gcloud services enable 
  run.googleapis.com 
  artifactregistry.googleapis.com 
  firebasehosting.googleapis.com 
  firebase.googleapis.com 
  iam.googleapis.com 
  cloudresourcemanager.googleapis.com

• Cloud Run API (run.googleapis.com) — deploy the FastAPI backend
• Artifact Registry API (artifactregistry.googleapis.com) — store container images
• Firebase Hosting API (firebasehosting.googleapis.com) — host the Angular frontend
• Firebase API (firebase.googleapis.com) — Firebase Anonymous Auth for the frontend
• IAM API (iam.googleapis.com) — manage service accounts and OIDC federation
• Cloud Resource Manager API (cloudresourcemanager.googleapis.com) — project-level operations

You’ll also want Docker installed locally (Dagger uses it under the hood), and Python 3.12+ and Node.js 20+ for the example apps themselves.

The Product: Angular Frontend + FastAPI Backend

Our product is two apps that talk to each other:

• Angular 21 SPA on Firebase Hosting: displays items from the API, authenticates via Firebase Anonymous Auth
• FastAPI REST API on Cloud Run: serves items, validates Firebase ID tokens from the frontend

The frontend authenticates anonymously with Firebase, gets an ID token, and sends it as a Bearer header to the backend. The backend validates the token using google-auth. This is a realistic pattern for any Firebase + Cloud Run stack.

Initializing the Module

Clone the companion repository:

git clone https://github.com/telchak/dagger-ci-demo.git
cd dagger-ci-demo

We have a monorepo with backend/ and frontend/. Rather than creating a separate Dagger module for each app, we’ll initialize a single module at the repository root. This gives us one place to define pipeline functions for the entire product:

dagger init --sdk=python --name=dagger-ci-demo

This creates a .dagger/ directory (Dagger’s convention, like .github/ for GitHub Actions) containing a dagger.json manifest, a pyproject.toml, and a source scaffold where we’ll write our pipeline functions.

Key Concepts: Types, Functions, and Chaining

Before looking at the code, let’s clarify three core Dagger concepts (docs):

Types. The Dagger API provides specialized types that represent CI/CD primitives. The ones we’ll use most:

• Container: an OCI container image with its filesystem and configuration
• Directory: a filesystem tree (your source code, build output, etc.)
• File: a single file artifact
• Secret: sensitive data (API keys, credentials) that Dagger never logs or caches
• Service: a running network service (for integration testing, local dev, etc.)

These types are the building blocks. They replace the implicit, untyped operations of YAML pipelines with explicit, typed abstractions.

Image generated with Google’s Gemini Imagen “Nano Banana Pro”

Functions. A Dagger module exposes functions, decorated with @function, that accept these types as parameters and return them as outputs. Functions are the unit of work. Each one is a self-contained operation: build a container, run tests, publish an image.

Chaining. Each Dagger type comes with its own methods, and these methods return the same (or related) types. This enables chaining: you pass the output of one operation directly into the next, building up a pipeline step by step. When you write dag.container().from_("python:3.12-slim").with_exec(["pip", "install", "..."]), you’re chaining three operations on a Container type. Dagger evaluates this chain lazily: nothing executes until a terminal operation (like .stdout() or .publish()) triggers the full graph.

The Pipeline: Backend and Frontend

Here’s what we’ll write at .dagger/src/dagger_ci_demo/main.py, a single class with functions for both apps:

"""CI/CD pipeline for the dagger-ci-demo product."""

from typing import Annotated

import dagger
from dagger import Doc, dag, function, object_type


# @object_type marks this class as a Dagger module.
# Its public methods become callable pipeline functions.
@object_type
class DaggerCiDemo:
    """Pipeline for the Angular + FastAPI product."""

    # --- Backend (FastAPI) ---

    @function
    async def build_backend(
        self,
        source: Annotated[dagger.Directory, Doc("Backend source directory")],
    ) -> dagger.Container:
        """Build the FastAPI backend container.

        Returns a Container — the core Dagger type representing an OCI image.
        The caller can publish it, export it, or chain further operations on it.
        """
        return (
            dag.container()
            # Start from a base image — returns a Container
            .from_("python:3.12-slim")
            # Set the working directory inside the container
            .with_workdir("/app")
            # Copy only requirements.txt first — this layer is cached
            # as long as requirements.txt doesn't change, even if
            # source code changes. This is the Docker cache optimization
            # pattern, expressed as a function chain.
            .with_file("/app/requirements.txt", source.file("requirements.txt"))
            # Install dependencies — cached if requirements.txt unchanged
            .with_exec(["pip", "install", "--no-cache-dir", "-r", "requirements.txt"])
            # Copy application source — only this layer and below re-run
            # when source code changes
            .with_directory("/app/src", source.directory("src"))
            # Configure the container for Cloud Run
            .with_env_variable("PORT", "8080")
            .with_exposed_port(8080)
            .with_entrypoint([
                "uvicorn", "src.main:app",
                "--host", "0.0.0.0", "--port", "8080", "--workers", "2",
            ])
        )

    @function
    async def test_backend(
        self,
        source: Annotated[dagger.Directory, Doc("Backend source directory")],
    ) -> str:
        """Run the backend test suite.

        Returns a string (stdout) — a terminal type that triggers execution.
        Everything before .stdout() is lazily evaluated; Dagger only runs
        the container when it needs the output.
        """
        return await (
            dag.container()
            .from_("python:3.12-slim")
            .with_workdir("/app")
            # Mount the full source directory (tests need everything)
            .with_directory("/app", source)
            .with_exec(["pip", "install", "--no-cache-dir", "-r", "requirements.txt"])
            # Run pytest — if this command fails, the function raises an error
            .with_exec(["pytest", "-v", "tests/"])
            # .stdout() is the terminal operation: it triggers execution
            # of the entire chain and returns the command's output as a string
            .stdout()
        )

    # --- Frontend (Angular) ---

    @function
    async def build_frontend(
        self,
        source: Annotated[dagger.Directory, Doc("Frontend source directory")],
    ) -> dagger.Directory:
        """Build the Angular frontend for production.

        Returns a Directory — the build output (dist/) that can be
        deployed to Firebase Hosting, exported to the host filesystem,
        or passed to another function.
        """
        return (
            dag.container()
            .from_("node:20-slim")
            .with_workdir("/app")
            # Copy package files first for dependency caching
            .with_file("/app/package.json", source.file("package.json"))
            .with_file("/app/package-lock.json", source.file("package-lock.json"))
            # npm ci installs exact versions from lockfile — cached if
            # package files unchanged
            .with_exec(["npm", "ci"])
            # Copy the rest of the source
            .with_directory("/app/src", source.directory("src"))
            .with_file("/app/angular.json", source.file("angular.json"))
            .with_file("/app/tsconfig.json", source.file("tsconfig.json"))
            .with_file("/app/tsconfig.app.json", source.file("tsconfig.app.json"))
            # Build for production — Angular CLI outputs to dist/
            .with_exec(["npx", "ng", "build", "--configuration=production"])
            # Extract just the build output directory.
            # .directory() returns a Directory type — stripping away the
            # container and keeping only the filesystem artifact we need.
            .directory("/app/dist/angular-frontend/browser")
        )

    @function
    async def test_frontend(
        self,
        source: Annotated[dagger.Directory, Doc("Frontend source directory")],
    ) -> str:
        """Run the Angular test suite."""
        return await (
            dag.container()
            .from_("node:20-slim")
            .with_workdir("/app")
            .with_directory("/app", source)
            .with_exec(["npm", "ci"])
            .with_exec(["npx", "ng", "test"])
            .stdout()
        )

Let’s unpack what’s going on here.

The class DaggerCiDemo is a module, a collection of functions packaged together. The @object_type decorator registers it with the Dagger engine, and every @function method becomes a callable operation, both from the CLI and from other modules.

Each function takes a Directory as input, Dagger’s type for a filesystem tree. When you pass --source=./backend from the CLI, Dagger packages that directory and hands it to the function as a typed Directory object. No string paths, no glob patterns, just a real, content-addressed filesystem reference.

The function bodies are chains of operations on Dagger types. dag.container().from_("python:3.12-slim") creates a Container from a base image. Each .with_*() call returns a new Container with the modification applied. Nothing executes yet. Dagger builds a directed acyclic graph (DAG) of operations. Execution only happens when a terminal operation like .stdout() or .publish() forces the engine to resolve the graph.

This is why caching works so well. Dagger hashes each node in the graph based on its inputs. If requirements.txt hasn’t changed, the pip install step is a cache hit, instantly. If only a source file changed, only the layers after .with_directory("/app/src", ...) re-execute. Same content-addressed logic, on your laptop and in CI.

Notice how build_backend returns a Container while build_frontend returns a Directory. The return type determines what the caller can do next. A Container can be published to a registry, started as a service, or chained further. A Directory can be deployed to static hosting, exported, or mounted into another container. Types guide composition. They make it clear what flows where.

Run It Locally

With Dagger installed (see prerequisites above), let’s explore the module from the repository root.

Discovering Functions

Before running anything, you can inspect what the module exposes:

dagger functions

Name             Description
build-backend    Build the FastAPI backend container.
build-frontend   Build the Angular frontend for production.
test-backend     Run the backend test suite.
test-frontend    Run the Angular test suite.

This is auto-generated. Dagger reads the @function decorators, their docstrings, and the Doc() annotations on parameters, and produces CLI documentation from them, in any SDK language. The same metadata that describes a function in Python generates the same CLI interface as it would in Go or TypeScript.

You can drill into any function with --help:

dagger call build-backend --help

Build the FastAPI backend container.

Returns a Container — the core Dagger type representing an OCI image.
The caller can publish it, export it, or chain further operations on it.

USAGE
  dagger call build-backend [arguments]

ARGUMENTS
  --source Directory   Backend source directory [required]

Everything here (the description, the argument name, the type, the [required] marker) comes directly from your code. The Doc("Backend source directory") annotation you wrote in the Annotated[dagger.Directory, Doc("...")] parameter becomes the argument’s help text. The function’s docstring becomes the command’s description. There’s no separate documentation to write or maintain. The code is the documentation, and Dagger keeps it in sync for both the CLI and the SDK.

This is what makes self-service work. A developer who has never seen this module before can run dagger functions, pick the operation they need, check its arguments with --help, and run it, without reading source code or asking the platform team.

Running Functions

# Test the backend
dagger call test-backend --source=./backend

# Build the backend container
dagger call build-backend --source=./backend

# Test the frontend
dagger call test-frontend --source=./frontend

# Build the frontend (returns the dist/ directory)
dagger call build-frontend --source=./frontend

A few things to notice about the CLI:

• dagger call is the universal entry point for running any function. It works the same way whether you’re calling a local module or a remote one (dagger call -m github.com/...).
• Function names are converted from Python’s snake_case to CLI kebab-case automatically (build_backend → build-backend).
• --source is not a built-in Dagger flag. It’s the parameter we defined with Annotated[dagger.Directory, Doc("...")]. Dagger turns every function parameter into a CLI flag, typed accordingly. A dagger.Directory parameter accepts a local path; a dagger.Secret would accept env:MY_SECRET or file:./key.json.
• Return types determine what the CLI shows. test_backend returns a str, so Dagger prints the stdout. build_backend returns a Container, so Dagger shows the image digest. build_frontend returns a Directory, which you can export with dagger call build-frontend --source=./frontend export --path=./dist.

Same functions, same commands, for both apps. A developer debugging a build failure doesn’t push a commit and wait. They run the same command on their laptop, inspect the output, and fix the issue with the same tools they use for application code. CI/CD is part of the software factory.

From Laptop to CI: The Same Commands in GitHub Actions

We’ve been running everything locally. Now let’s prove the claim: the same code runs in CI without any changes. Create .github/workflows/ci.yml:

name: CI

on:
  push:
    branches: [main]
  pull_request:

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v6

      - name: Test Backend
        uses: dagger/[email protected]
        with:
          version: "0.20.3"
          verb: call
          args: test-backend --source=./backend

      - name: Test Frontend
        uses: dagger/[email protected]
        with:
          version: "0.20.3"
          verb: call
          args: test-frontend --source=./frontend

That’s it. No Dockerfile to maintain. No shell scripts gluing steps together. No CI-specific logic. The dagger/dagger-for-github action installs the Dagger CLI and engine, and runs the exact same dagger call commands we just ran locally. The pipeline code doesn’t know, or care, where it’s running.

This is the minimal, working CI setup. Push it, open a pull request, and watch the same tests you just ran on your laptop execute in GitHub Actions. Same functions, same caching, same results.

Why Now

Three trends make this approach timely:

Platform Engineering is maturing. The core promise of Platform Engineering is developer self-service: give teams golden paths so they can ship without waiting for specialists. CI/CD modules are the missing piece in that story: reusable, versioned, self-documenting abstractions that reduce cognitive load the same way a Terraform module does. When CI/CD is software, it fits naturally into the Internal Developer Platform.

AI agents need programmatic interfaces. When your pipeline is real code with typed functions, AI agents can interact with it through tool use. YAML pipelines are opaque to agents. Typed Dagger functions are not. (We’ll explore this in Part 4.)

Multi-cloud is the norm. A pipeline that deploys to GCP today can deploy to AWS tomorrow with a module swap. No vendor lock-in at the CI layer. The platform team maintains the modules; developers maintain their code.

What’s Next

We now have a working CI pipeline, real code running both locally and in GitHub Actions with the same commands. But this minimal setup has limitations: every job starts with a cold cache, builds run on GitHub’s shared 2-vCPU runners, and there’s no persistent state between runs.

In Part 2, we’ll tackle the infrastructure side: caching strategies that cut build times by 5x, managed runners with persistent NVMe storage, and self-hosted Kubernetes runners with shared Dagger engines. Three approaches, each with different tradeoffs between simplicity and control.

The pipeline code won’t change. Only where it runs will.

Up Next: Part 2: Decoupling Pipelines from Infrastructure

This is Part 1 of a 4-part series. Follow for updates.

Tags: #cicd #dagger #platform-engineering #python #angular

I opened a new Claude Code session and asked it to add a feature to a tool I’d been building for three weeks. It asked me what language the project was in.

Three weeks. Dozens of sessions. Hundreds of decisions. And the AI started from zero — as if we’d never met.

That’s not a bug. That’s how every AI coding agent works. Claude Code, Cursor, Windsurf, Copilot — every session is a blank slate. The agent can write complex systems in minutes, but it can’t remember what you named the database table yesterday.

Most people solve this by re-explaining. They paste context into the first message. They describe the project. They remind the AI about conventions. Every single session.

Then they wonder why they’re burning through their Max plan 5x allowance in two hours.

Every re-explanation is tokens. Every clarifying question the AI asks because it doesn’t know your stack — tokens. Every time it suggests a migration you already rejected — tokens, plus the tokens you spend correcting it. You’re not paying for work. You’re paying for the AI to catch up to where it was yesterday.

There’s a file that fixes this. And most people either don’t have one, or they have one that makes things worse.

The Three Ways It Goes Wrong

After 60+ sessions building production tools, I’ve seen the same three mistakes — in my own projects and in every public repo I’ve checked.

1. The Empty File

No context file at all. Or something like:

# My Project
A web app built with Flask.

The AI knows nothing. It asks what framework you’re using — again. It suggests PostgreSQL when you already chose SQLite — again. Every clarifying question eats into the context window you need for actual work.

2. The Kitchen Sink

The opposite extreme: 400 lines. Full README. Changelog. API docs. Dependency lists. Three paragraphs of project philosophy. Installation instructions the AI will never use.

This costs money directly. That file loads into the context window at session start. On API pricing, 400 lines is roughly 8,000 tokens — loaded before you’ve typed a word. Across 50 sessions, that’s 400,000 tokens of instructions the AI re-reads every time. Most of it noise.

The AI doesn’t need your API documentation. It can read the code. It needs your operating context — the decisions, patterns, and gotchas that aren’t in the code.

3. The Stale Snapshot

A context file written two months ago that still references the database you migrated away from and the deployment target you abandoned.

This is the sneakiest failure mode. The AI generates code that looks correct but targets the wrong architecture. You don’t catch it until the tests fail — or worse, until production breaks.

Stale context is worse than no context. No context makes the AI ask questions. Stale context makes it confident and wrong.

What Actually Works

The fix isn’t “write a better prompt.” It’s context engineering — giving the AI the right operating context, compressed, current, and curated.

A working context file has seven sections:

1. Project Summary — 2-3 lines. Stack, purpose, deployment target. Not what it could be. What it is right now.
2. Architecture — One bullet per component. File name, what it does, what it connects to.
3. Key Decisions — Things you’ve already decided. The AI should not relitigate these.
4. Patterns & Conventions — How things work here. Not rules — descriptions of existing patterns.
5. Gotchas & Pitfalls — Every gotcha entry is an hour of debugging you’ll never repeat.
6. Current Progress — Where you are right now. Not where you were two months ago.
7. Context — Hardware, environment, constraints the AI can’t infer from code.

Each section is compressed bullets with tags. The whole file stays under 100 lines. Everything loads fast, everything earns its place.

The difference between a project with this file and one without it is the difference between onboarding a new hire every morning and working with someone who was there yesterday.

The Playbook

I wrote the full system into a playbook — 8 chapters covering structure, cost discipline, security patterns, production gotchas, advanced features, and before/after transformations. Plus 6 copy-paste starter templates for different stacks (Python/Flask, Next.js, Rust, monorepo, Docker, solo dev).

It’s built from 60+ real sessions shipping real tools, not theory.

If you’re tired of re-explaining your project to your AI every session, this is the file that fixes it.

The CLAUDE.md Playbook — $12 on LemonSqueezy

Built by WebsationFlow. Zero-cost AI tools from consumer hardware.

5 AI Agent Memory Patterns That Actually Work (With Python Code)

Every AI agent starts stateless. Each request is a blank slate. That works for simple tasks, but the moment you need context — “what did the user ask yesterday?” or “what files did I already review?” — you need memory.

Here are 5 memory patterns, ordered from simplest to most powerful. Each one includes working code.

Pattern 1: Conversation Buffer (The Default)

Most frameworks give you this for free. Store the full conversation history and pass it back every time.

class BufferMemory:
    def __init__(self, max_messages: int = 50):
        self.messages: list[dict] = []
        self.max_messages = max_messages

    def add(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        # Trim from the front when we hit the limit
        if len(self.messages) > self.max_messages:
            self.messages = self.messages[-self.max_messages:]

    def get_context(self) -> list[dict]:
        return self.messages.copy()

When to use: Simple chatbots, single-session tools.

The problem: Token cost scales linearly. At 50 messages, you’re burning thousands of tokens per call just on context. And when the session ends, everything disappears.

Pattern 2: Sliding Window + Summary

Keep recent messages verbatim, but compress older ones into a running summary.

import json

class SlidingWindowMemory:
    def __init__(self, window_size: int = 10):
        self.recent: list[dict] = []
        self.summary: str = ""
        self.window_size = window_size

    def add(self, role: str, content: str):
        self.recent.append({"role": role, "content": content})
        if len(self.recent) > self.window_size:
            overflow = self.recent[:-self.window_size]
            self.recent = self.recent[-self.window_size:]
            self._update_summary(overflow)

    def _update_summary(self, old_messages: list[dict]):
        # In production, use an LLM call here
        conversation = "n".join(
            f"{m['role']}: {m['content']}" for m in old_messages
        )
        # Simplified — replace with actual LLM summarization
        self.summary += f"n[Summarized: {len(old_messages)} messages about "
        self.summary += f"{old_messages[0]['content'][:50]}...]"

    def get_context(self) -> list[dict]:
        context = []
        if self.summary:
            context.append({
                "role": "system",
                "content": f"Previous conversation summary:n{self.summary}"
            })
        context.extend(self.recent)
        return context

When to use: Long conversations, customer support agents, coding assistants.

The key insight: Summaries are lossy. You trade precision for cost. The last 10 messages are exact; older ones are compressed. For most agents, this is the right tradeoff.

Pattern 3: File-Based Persistent Memory

The simplest form of persistence. Write memories to disk, load them next session.

import json
from pathlib import Path
from datetime import datetime

class FileMemory:
    def __init__(self, memory_dir: str = "./memory"):
        self.dir = Path(memory_dir)
        self.dir.mkdir(exist_ok=True)

    def save_daily(self, content: str):
        """Save a daily memory note."""
        today = datetime.now().strftime("%Y-%m-%d")
        path = self.dir / f"{today}.md"
        with open(path, "a") as f:
            timestamp = datetime.now().strftime("%H:%M")
            f.write(f"n## {timestamp}n{content}n")

    def save_entity(self, entity: str, facts: dict):
        """Save facts about a specific entity (user, project, etc)."""
        path = self.dir / "entities.json"
        entities = {}
        if path.exists():
            entities = json.loads(path.read_text())

        if entity not in entities:
            entities[entity] = {}
        entities[entity].update(facts)
        entities[entity]["_updated"] = datetime.now().isoformat()

        path.write_text(json.dumps(entities, indent=2))

    def load_recent(self, days: int = 3) -> str:
        """Load memory from the last N days."""
        memories = []
        for i in range(days):
            date = datetime.now() - timedelta(days=i)
            path = self.dir / f"{date.strftime('%Y-%m-%d')}.md"
            if path.exists():
                memories.append(path.read_text())
        return "n---n".join(memories)

    def search(self, keyword: str) -> list[str]:
        """Simple keyword search across all memory files."""
        results = []
        for path in self.dir.glob("*.md"):
            content = path.read_text()
            if keyword.lower() in content.lower():
                results.append(f"**{path.name}:**n{content}")
        return results

When to use: Personal assistants, project-specific agents, any agent that needs to “remember” across sessions.

Why this works: Files are debuggable. You can read them, edit them, version-control them. No database setup, no dependencies. This is how many production AI assistants actually work.

Pattern 4: Structured Memory with Decay

Not all memories are equal. Recent ones matter more. Frequently-accessed ones matter more. This pattern adds relevance scoring.

from dataclasses import dataclass, field
from datetime import datetime, timedelta
import math
import json

@dataclass
class Memory:
    content: str
    created: datetime = field(default_factory=datetime.now)
    last_accessed: datetime = field(default_factory=datetime.now)
    access_count: int = 0
    importance: float = 0.5  # 0-1 scale

    def relevance_score(self) -> float:
        """Combine recency, frequency, and importance."""
        hours_since_access = (
            datetime.now() - self.last_accessed
        ).total_seconds() / 3600

        # Exponential decay: half-life of 24 hours
        recency = math.exp(-0.693 * hours_since_access / 24)

        # Log-scale frequency boost
        frequency = math.log(self.access_count + 1) / 5

        # Weighted combination
        return (0.4 * recency) + (0.2 * frequency) + (0.4 * self.importance)


class DecayMemory:
    def __init__(self, max_memories: int = 1000):
        self.memories: list[Memory] = []
        self.max_memories = max_memories

    def add(self, content: str, importance: float = 0.5):
        mem = Memory(content=content, importance=importance)
        self.memories.append(mem)
        self._prune()

    def recall(self, top_k: int = 10) -> list[str]:
        """Return the most relevant memories."""
        scored = sorted(
            self.memories,
            key=lambda m: m.relevance_score(),
            reverse=True
        )[:top_k]

        # Mark as accessed
        for m in scored:
            m.last_accessed = datetime.now()
            m.access_count += 1

        return [m.content for m in scored]

    def _prune(self):
        """Remove lowest-relevance memories when over capacity."""
        if len(self.memories) > self.max_memories:
            self.memories.sort(key=lambda m: m.relevance_score())
            self.memories = self.memories[-self.max_memories:]

When to use: Long-running agents that accumulate thousands of memories. Agents that need to “forget” gracefully.

The math: The relevance score combines three signals — recency (exponential decay with 24h half-life), access frequency (log-scaled), and importance (set when memory is created). Tweak the weights for your use case.

Pattern 5: Vector-Backed Semantic Memory

When keyword search isn’t enough. Store memories as embeddings, retrieve by meaning.

import numpy as np
from openai import OpenAI

class VectorMemory:
    def __init__(self):
        self.client = OpenAI()
        self.memories: list[dict] = []  # {text, embedding, metadata}

    def _embed(self, text: str) -> list[float]:
        response = self.client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding

    def add(self, text: str, metadata: dict = None):
        embedding = self._embed(text)
        self.memories.append({
            "text": text,
            "embedding": embedding,
            "metadata": metadata or {},
            "created": datetime.now().isoformat(),
        })

    def search(self, query: str, top_k: int = 5) -> list[str]:
        """Find memories most similar to the query."""
        query_embedding = self._embed(query)

        scored = []
        for mem in self.memories:
            similarity = self._cosine_similarity(
                query_embedding, mem["embedding"]
            )
            scored.append((similarity, mem["text"]))

        scored.sort(reverse=True)
        return [text for _, text in scored[:top_k]]

    @staticmethod
    def _cosine_similarity(a: list[float], b: list[float]) -> float:
        a, b = np.array(a), np.array(b)
        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

When to use: Knowledge bases, RAG systems, agents that need to find relevant info across thousands of unstructured memories.

Production tip: For anything beyond 10K memories, use a vector database (Chroma, Pinecone, pgvector) instead of in-memory cosine similarity. The pattern stays the same — only the storage layer changes.

Choosing the Right Pattern

Start with Pattern 3 (file-based). It’s simple, debuggable, and handles most use cases. Add vector search only when keyword matching fails.

The Hybrid Approach

Production agents typically combine patterns:

class HybridMemory:
    def __init__(self):
        self.buffer = SlidingWindowMemory(window_size=10)
        self.files = FileMemory("./memory")
        self.vectors = VectorMemory()

    def get_context(self, query: str = None) -> list[dict]:
        context = []

        # Layer 1: Recent conversation (always included)
        context.extend(self.buffer.get_context())

        # Layer 2: Today's notes (file-based)
        daily = self.files.load_recent(days=1)
        if daily:
            context.append({
                "role": "system",
                "content": f"Today's notes:n{daily}"
            })

        # Layer 3: Semantic recall (if we have a query)
        if query:
            relevant = self.vectors.search(query, top_k=3)
            if relevant:
                context.append({
                    "role": "system",
                    "content": f"Relevant memories:n" + 
                              "n".join(f"- {m}" for m in relevant)
                })

        return context

Short-term buffer for the current conversation. Files for structured daily/entity data. Vectors for semantic search across everything.

Key Takeaways

1. Stateless is a bug, not a feature — agents without memory repeat mistakes and lose context
2. Files beat databases for most agent memory needs — simpler, more debuggable
3. Decay matters — not all memories deserve equal weight
4. Hybrid wins — combine patterns for different time horizons
5. Start simple — you can always add vector search later

This is part of the “AI Engineering in Practice” series — practical guides for developers building with AI. Follow for more.

The first disposable email service launched in 2000. A developer named Josh built Spamgourmet — you created addresses like [email protected], where the 3 meant “accept three emails then stop forwarding.” The counter was baked into the address itself. No web UI needed. Josh maintained it until he died of brain cancer in 2020. His son Josiah runs it now.

Twenty-six years later, AI agents are creating their own email inboxes without human intervention. The throughline between these two things is shorter than you’d think.

Era 1: Anti-Spam (2000-2010)

By the late 1990s, every website wanted your email to register. Every registration meant more spam. The CAN-SPAM Act wouldn’t arrive until 2003. People needed a way to give out email addresses that worked but didn’t matter.

Mailinator (2003) defined the category. Paul Tyma, a Google engineer, got the idea from his drunk roommate. The first version took three days to code. The concept: any email sent to [email protected] creates an inbox. No signup. No password. Just visit the site, type the address, and read whatever arrived.

The original ran on a single server — AMD 2GHz Athlon, 1GB RAM, 80GB IDE hard drive. It processed 4.5 million emails a day. Tyma’s architecture blog post went viral on Hacker News: keep everything in RAM, delete oldest first, never touch disk. The entire stack — web app, SMTP server, storage — ran in a single JVM. The Washington Post and New York Times wrote about it.

10 Minute Mail (2006) added the countdown timer. Devon Hillard built it as a weekend project to learn JBoss Seam. The UX innovation was the constraint — your address self-destructs in 10 minutes. No choices to make. Visit the site, get an address, use it, walk away. It spawned an entire genre: 20 Minute Mail, 30 Minute Mail, Minute Inbox.

Guerrilla Mail (2006) came out of Chicago and quietly became a machine. Twenty billion emails processed over its lifetime. A hundred and thirty thousand emails per hour, currently. The team built their own SMTP server in Go — open-sourced as Go-Guerrilla — because existing mail servers were too bloated. Guerrilla Mail was also the first to let you send from a disposable address, not just receive.

YOPmail (2004) won France. TrashMail (2002) won Germany. Discardmail (2004) survived two decades under four different names. Each served the same purpose: a real email address for a task you didn’t care about keeping.

The users were humans. The interface was a browser. The business model was free. The enemy was spam.

The Blocklist Wars

As disposable email services multiplied, websites fought back.

Websites built blocklists of known disposable domains. Services created alternate domains — Mailinator alone has hundreds, including gems like thisisnotmyrealemail.com. Open-source blocklists appeared on GitHub around 2014, community-maintained and updated daily. By 2025, they tracked over 110,000 disposable email domains.

The cat-and-mouse game escalated. Services rotated domains. Users pointed their own custom MX records at Mailinator — completely undetectable by blocklists. Detection APIs emerged — IPQualityScore, UserCheck — offering real-time disposable email identification as a service.

Temp-mail.org won this war through distribution, not technology. The domain was nearly perfect for search intent. Available in 29 languages. Chrome extension. Android app. 56% of their 15 million monthly visits come from organic search. They didn’t build the best disposable email service. They built the most findable one.

Era 2: Software Testing (2011-2024)

Around 2011, the use case shifted. Disposable email stopped being about avoiding spam and started being about testing software.

Guerrilla Mail’s API (April 2011) was the first. Free, public, no registration — developers could create and check disposable inboxes programmatically. This was the moment disposable email became infrastructure instead of a consumer tool.

Mailsac (2012) and Mailosaur (2013) were built specifically for QA teams. The founders needed to test their own email workflows, built their own tools, and opened them up. These integrated with Selenium, Playwright, Cypress — the browser automation frameworks that QA teams were already using.

Mailinator pivoted. What started as a free consumer tool became an enterprise email testing platform. Private domains, REST API, webhooks, SMS testing (added in 2018), email load testing. Pricing went from free to $79-$459 per month. Revenue hit $126K with 300 enterprise customers by 2017, bootstrapped.

The customer was no longer a person avoiding spam. It was a QA team running email tests in CI/CD pipelines. The interface was an API, not a browser. The business model was SaaS.

Meanwhile, Big Tech legitimized email aliasing for consumers. Apple launched Hide My Email in 2021 — unlimited random addresses forwarding to your real inbox, $0.99/month with iCloud+. Mozilla launched Firefox Relay. Proton acquired SimpleLogin in 2022. The concept Spamgourmet pioneered in 2000 became a first-party feature in every major ecosystem.

By 2024, the disposable email market was valued at $1.36 billion, growing at 11% annually, projected to hit $3.53 billion by 2033.

Era 3: AI Agents (2025-present)

The user changed again. It’s no longer a human avoiding spam or a QA team testing workflows. It’s an autonomous AI agent that needs an email address to complete a task.

The trigger was OpenClaw launching in late January 2026. Suddenly, hundreds of thousands of persistent AI agents were running on VPSes, browsing the web, signing up for services, and hitting the same wall: registration forms that require email verification.

AgentMail raised $6 million in March 2026 to build persistent email identities for agents. Founded by an ex-Optiver quant, an ex-NVIDIA autonomous vehicles engineer, and an ex-Accel investor. General Catalyst led the round. Paul Graham and Dharmesh Shah (HubSpot CTO) are angels. Their user count tripled when OpenClaw launched, quadrupled in February.

The defining moment from their blog: autonomous agents started signing up for AgentMail on their own — finding it through web search, navigating to the site, and creating inboxes without a developer in the loop. AgentMail built agent.email — a landing page designed not for humans but for agents to self-provision.

KeyID took a different approach — free email, phone, and SMS infrastructure for agents. Ed25519 keypair auth instead of API keys. MCP server with 47 tools. Free at 1,000 accounts.

Agent Burner is essentially temp-mail.org for agents — no account, no API key, just HTTP. A single POST creates a disposable inbox that auto-expires in an hour. URLs are pre-extracted from incoming emails so the agent doesn’t parse HTML. The same need Mailinator solved for humans in 2003, but now the user types curl instead of visiting a website.

The throughline

Every era of disposable email solves the same problem for a different user.

2003: A human needs to register for a website without getting spammed. They visit Mailinator in a browser and type an address.

2013: A QA team needs to test a signup flow in CI. They call an API to create an inbox, run the test, and check for the verification email programmatically.

2026: An autonomous agent needs to sign up for a service at 3am without human intervention. It creates an inbox via HTTP, fills a registration form in Turkish, receives a verification email, extracts the link, and deletes the inbox. The address existed for four minutes.

The throughline: something needs a real email address for a task, but doesn’t want to keep it. The technology barely changed. A disposable email service in 2026 does the same thing Mailinator did in 2003 — receive an email and let someone read it. What changed is who’s reading it, and why.

Paul Tyma built Mailinator because his drunk roommate had an idea. Josh built Spamgourmet and maintained it until he couldn’t anymore. Devon Hillard built 10 Minute Mail to learn a framework. The Go-Guerrilla team built their own SMTP server because Sendmail was too bloated. Guerrilla Mail has processed twenty billion emails without venture funding.

Disposable email is a category that keeps getting reinvented because the problem never goes away. Someone always needs an address they don’t plan to keep.

The AI landscape is experiencing unprecedented growth and transformation. This post delves into the key developments shaping the future of artificial intelligence, from massive industry investments to critical safety considerations and integration into core development processes.

Key Areas Explored:

• Record-Breaking Investments: Major tech firms are committing billions to AI infrastructure, signaling a significant acceleration in the field.
• AI in Software Development: We examine how companies are leveraging AI for code generation and the implications for engineering workflows.
• Safety and Responsibility: The increasing focus on ethical AI development and protecting vulnerable users, particularly minors.
• Market Dynamics: How AI is influencing stock performance, cloud computing strategies, and global market trends.
• Global AI Strategies: Companies are adapting AI development for specific regional markets.

This deep dive aims to provide developers, tech leaders, and enthusiasts with a comprehensive overview of the current state and future trajectory of AI.

AI #ArtificialIntelligence #TechTrends #SoftwareEngineering #MachineLearning #CloudComputing #FutureOfTech #AISafety