Building a different app on Amplifier

The CLI is one application. You can build a completely different interface (for example, a web app that runs multiple tasks) by hosting the same kernel contracts and loading the same kinds of modules.

Target example: a web app that runs multiple tasks

Imagine a server that accepts “task requests” (each with a prompt, profile, and constraints), runs them concurrently, and streams progress + final results to a browser UI.

Recommended architecture (conceptual)

Key design decisions you’ll make

• Isolation: per-task sandboxing, filesystem scope, and credentials boundaries.
• Concurrency model: one process per task vs async coroutines vs worker queue.
• Event transport: SSE/WebSockets; you can treat the kernel’s event stream as the canonical source.
• Profiles as presets: let users select “dev” vs “foundation” vs “test” per task.
• Approvals: web UI equivalents of interactive approvals (human-in-the-loop gates).

How you would reuse the ecosystem

• Reuse provider modules to avoid re-implementing model integrations.
• Reuse tool modules where safe (filesystem/bash might need stronger sandboxing in a web server).
• Reuse hooks for logging/redaction, but route output into your web observability stack.
• Reuse profiles (or author new ones) as user-facing “modes”.

Real Example: AI Email Assistant Web App

A complete example showing how to build a web application that uses Amplifier to intelligently handle emails.

Requirements

• Read emails from Gmail (OAuth) or IMAP
• Create one Amplifier session per email (deterministic session ID)
• Auto-refresh emails every minute
• Web UI to configure, view emails, and trigger AI actions
• Stream AI responses in real-time
• Actions: reply, archive, custom workflows

Architecture Overview

Technology Stack

Step 1: Email Service (IMAP Example)

# email_service.py
import imaplib
import email
from email.header import decode_header
from datetime import datetime
import hashlib
import sqlite3

class EmailManager:
    def __init__(self, db_path="data/emails.db"):
        self.conn = sqlite3.connect(db_path)
        self._init_db()
        self.imap = None

    def _init_db(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS emails (
                id TEXT PRIMARY KEY,
                sender TEXT,
                subject TEXT,
                date TEXT,
                body TEXT,
                seen BOOLEAN DEFAULT 0,
                handled BOOLEAN DEFAULT 0
            )
        """)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS config (
                key TEXT PRIMARY KEY,
                value TEXT
            )
        """)
        self.conn.commit()

    def connect_imap(self, server, email_addr, password):
        """Connect to IMAP server and save config"""
        self.imap = imaplib.IMAP4_SSL(server)
        self.imap.login(email_addr, password)

        # Save config (encrypt in production!)
        self.conn.execute(
            "INSERT OR REPLACE INTO config VALUES (?, ?)",
            ("imap_server", server)
        )
        self.conn.execute(
            "INSERT OR REPLACE INTO config VALUES (?, ?)",
            ("email", email_addr)
        )
        # In production: use keyring or encrypted storage for password
        self.conn.commit()

    def deterministic_id(self, sender, subject, date):
        """Generate deterministic session ID"""
        content = f"{sender}|{subject}|{date}"
        return hashlib.sha256(content.encode()).hexdigest()[:16]

    def fetch_new_emails(self):
        """Fetch new emails and store in DB"""
        if not self.imap:
            raise Exception("Not connected to IMAP")

        self.imap.select("INBOX")
        _, messages = self.imap.search(None, "UNSEEN")

        new_emails = []
        for num in messages[0].split():
            _, msg_data = self.imap.fetch(num, "(RFC822)")
            email_body = msg_data[0][1]
            email_message = email.message_from_bytes(email_body)

            sender = email_message.get("From", "")
            subject = email_message.get("Subject", "")
            date = email_message.get("Date", "")

            # Decode subject if needed
            if subject:
                decoded = decode_header(subject)[0]
                if isinstance(decoded[0], bytes):
                    subject = decoded[0].decode(decoded[1] or 'utf-8')

            # Get email body
            body = ""
            if email_message.is_multipart():
                for part in email_message.walk():
                    if part.get_content_type() == "text/plain":
                        body = part.get_payload(decode=True).decode()
                        break
            else:
                body = email_message.get_payload(decode=True).decode()

            email_id = self.deterministic_id(sender, subject, date)

            # Store in DB
            self.conn.execute("""
                INSERT OR IGNORE INTO emails
                (id, sender, subject, date, body)
                VALUES (?, ?, ?, ?, ?)
            """, (email_id, sender, subject, date, body))

            new_emails.append({
                "id": email_id,
                "sender": sender,
                "subject": subject,
                "date": date,
                "body": body
            })

        self.conn.commit()
        return new_emails

    def get_all_emails(self):
        """Get all emails from DB"""
        cursor = self.conn.execute(
            "SELECT id, sender, subject, date, handled FROM emails ORDER BY date DESC"
        )
        return [
            {"id": r[0], "sender": r[1], "subject": r[2], "date": r[3], "handled": bool(r[4])}
            for r in cursor.fetchall()
        ]

    def get_email(self, email_id):
        """Get email details"""
        cursor = self.conn.execute(
            "SELECT sender, subject, date, body FROM emails WHERE id = ?",
            (email_id,)
        )
        row = cursor.fetchone()
        if not row:
            return None
        return {
            "id": email_id,
            "sender": row[0],
            "subject": row[1],
            "date": row[2],
            "body": row[3]
        }

    def mark_handled(self, email_id):
        """Mark email as handled"""
        self.conn.execute(
            "UPDATE emails SET handled = 1 WHERE id = ?",
            (email_id,)
        )
        self.conn.commit()

Step 2: Email Tool Module

Create a proper Amplifier module at modules/tool-email/. This allows the tool to be:

• Loaded through the bundle (not hardcoded)
• Accessed via coordinator (shared instance)
• Potentially published to git for reuse

See examples/email-assistant/modules/tool-email/ for the complete implementation, or Creating Custom Modules guide.

Key files:

• pyproject.toml - Module metadata with entry point
• amplifier_module_tool_email/__init__.py - mount() function
• amplifier_module_tool_email/manager.py - EmailManager (IMAP, SQLite)
• amplifier_module_tool_email/tool.py - EmailTool with operations

The tool provides operations:

# Operation-based tool (like tool-issue)
async def execute(self, input: dict) -> ToolResult:
    operation = input.get("operation")
    params = input.get("params", {})

    if operation == "reply":
        return await self._reply(params)
    elif operation == "archive":
        return await self._archive(params)
    elif operation == "fetch":
        return await self._fetch(params)
    # ... more operations

Step 3: Amplifier Integration

# amplifier_service.py
from amplifier_core import AmplifierSession
from amplifier_foundation.bundle import load_bundle
from email_tools import EmailReplyTool, EmailArchiveTool
import asyncio

class AmplifierEmailService:
    def __init__(self):
        self.sessions = {}  # session_id -> AmplifierSession
        self.prepared_bundle = None

    async def initialize(self):
        """Initialize async (bundle loading is async)."""
        bundle_path = Path(__file__).parent.parent / "bundle.md"
        bundle = await load_bundle(str(bundle_path))
        self.prepared_bundle = await bundle.prepare()  # Load and validate modules

    async def handle_email(self, email_id: str):
        """Create session and handle email with AI"""

        # Get email details
        email_data = self.email_service.get_email(email_id)
        if not email_data:
            raise ValueError(f"Email {email_id} not found")

        # Use deterministic session ID
        session_id = email_id

        # Create or resume session
        if session_id not in self.sessions:
            # Add email-specific tools to mount plan
            tools = self.mount_plan["tools"].copy()
            tools.append(EmailReplyTool(self.email_service, email_id))
            tools.append(EmailArchiveTool(self.email_service, email_id))

            # Create session
            session = AmplifierSession(
                session_id=session_id,
                orchestrator=self.mount_plan["orchestrator"],
                providers=self.mount_plan["providers"],
                tools=tools,
                hooks=self.mount_plan["hooks"],
                context=self.mount_plan["context"]
            )

            self.sessions[session_id] = session
        else:
            session = self.sessions[session_id]

        # Compose prompt for AI
        prompt = f"""
You are handling an email. Here are the details:

From: {email_data['sender']}
Subject: {email_data['subject']}
Date: {email_data['date']}

Body:
{email_data['body']}

---

Please analyze this email and suggest appropriate actions. You can:
1. Compose a reply using the email_reply tool
2. Archive it using the email_archive tool
3. Ask me for guidance if unsure

What should we do with this email?
"""

        # Run session and yield events
        async for event in session.run_async(prompt):
            yield event

Step 4: FastAPI Backend

# main.py - FastAPI application
from fastapi import FastAPI, BackgroundTasks
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import asyncio
import json
from amplifier_service import AmplifierEmailService
from amplifier_core import AmplifierSession

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # Your frontend
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Services
amplifier_svc = None
email_tool = None  # Will be set after bundle loads

# Background task to poll emails
async def poll_emails():
    while True:
        try:
            new_emails = email_svc.fetch_new_emails()
            print(f"Fetched {len(new_emails)} new emails")
        except Exception as e:
            print(f"Error fetching emails: {e}")
        await asyncio.sleep(60)  # Every minute

@app.on_event("startup")
async def startup():
    global amplifier_svc, email_tool

    # Initialize Amplifier service
    amplifier_svc = AmplifierEmailService()
    await amplifier_svc.initialize()

    # Get email tool from coordinator
    temp_session = await amplifier_svc.prepared_bundle.create_session(
        session_id="temp-for-tool-access"
    )
    email_tool = temp_session.coordinator.get("tools", "email")

    # Start background email polling
    asyncio.create_task(poll_emails())

# API Models
class IMAPConfig(BaseModel):
    server: str
    email: str
    password: str

class EmailAction(BaseModel):
    action: str  # "handle", "archive", "reply"
    prompt: str | None = None

# API Endpoints
@app.post("/api/config/imap")
async def configure_imap(config: IMAPConfig):
    """Configure IMAP connection"""
    try:
        email_svc.connect_imap(config.server, config.email, config.password)
        return {"status": "connected"}
    except Exception as e:
        return {"error": str(e)}, 400

@app.get("/api/emails")
async def list_emails():
    """List all emails"""
    emails = email_svc.get_all_emails()
    return {"emails": emails}

@app.get("/api/emails/{email_id}")
async def get_email(email_id: str):
    """Get email details"""
    email_data = email_svc.get_email(email_id)
    if not email_data:
        return {"error": "Email not found"}, 404
    return email_data

@app.post("/api/emails/{email_id}/handle")
async def handle_email(email_id: str, action: EmailAction):
    """Handle email with AI (Server-Sent Events stream)"""

    async def event_stream():
        try:
            async for event in amplifier_svc.handle_email(email_id):
                # Stream events to frontend
                yield f"data: {json.dumps(event)}\n\n"
        except Exception as e:
            yield f"data: {json.dumps({'error': str(e)})}\n\n"

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream"
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Step 5: Email Assistant Bundle

# bundle.md (at project root)
---
bundle:
  name: email-assistant
  version: 1.0.0
  description: AI email assistant with reply/archive capabilities

includes:
  - bundle: git+https://github.com/microsoft/amplifier-foundation@main

providers:
  - module: provider-anthropic
    source: git+https://github.com/microsoft/amplifier-module-provider-anthropic@main
    config:
      default_model: claude-sonnet-4-5

tools:
  - module: tool-email
    source: ./modules/tool-email  # Local module
    config:
      db_path: ../data/emails.db  # Relative to module location

hooks:
  - module: hooks-logging
    source: git+https://github.com/microsoft/amplifier-module-hooks-logging@main
---

# Email Assistant

You are an intelligent email assistant.

## Your Tools

Use the **email** tool with these operations:
- email(operation="get", params={"email_id": "..."})
- email(operation="reply", params={"email_id": "...", "reply_text": "...", "action": "send"})
- email(operation="archive", params={"email_id": "..."})
- email(operation="label", params={"email_id": "...", "label": "urgent"})

## Your Approach

1. Analyze the email (sender, subject, urgency)
2. Determine action (reply/archive/label/ask)
3. Use the email tool to execute actions
4. Ask user if unsure

Step 6: Frontend (React Example)

// EmailList.jsx
import { useState, useEffect } from 'react';

function EmailList() {
  const [emails, setEmails] = useState([]);
  const [selectedEmail, setSelectedEmail] = useState(null);
  const [aiResponse, setAiResponse] = useState("");

  // Fetch emails every 5 seconds
  useEffect(() => {
    const fetchEmails = async () => {
      const res = await fetch('http://localhost:8000/api/emails');
      const data = await res.json();
      setEmails(data.emails);
    };

    fetchEmails();
    const interval = setInterval(fetchEmails, 5000);
    return () => clearInterval(interval);
  }, []);

  const handleWithAI = async (emailId) => {
    setAiResponse("Processing...");

    // Open SSE connection
    const eventSource = new EventSource(
      `http://localhost:8000/api/emails/${emailId}/handle`
    );

    eventSource.onmessage = (event) => {
      const data = JSON.parse(event.data);

      if (data.type === 'text') {
        setAiResponse(prev => prev + data.content);
      } else if (data.type === 'tool_result') {
        setAiResponse(prev => prev + `\n[Tool: ${data.tool}] ${data.output}`);
      } else if (data.type === 'complete') {
        eventSource.close();
      }
    };

    eventSource.onerror = () => {
      eventSource.close();
      setAiResponse(prev => prev + "\n[Error: Connection lost]");
    };
  };

  return (
    <div>
      <h1>AI Email Assistant</h1>

      <div className="email-list">
        {emails.map(email => (
          <div key={email.id} className="email-item">
            <div>
              <strong>{email.subject}</strong>
              <div>From: {email.sender}</div>
            </div>
            <button onClick={() => handleWithAI(email.id)}>
              Handle with AI
            </button>
            {email.handled && <span>✓ Handled</span>}
          </div>
        ))}
      </div>

      {aiResponse && (
        <div className="ai-response">
          <h2>AI Response</h2>
          <pre>{aiResponse}</pre>
        </div>
      )}
    </div>
  );
}

Step 7: Deployment

# Install dependencies
uv venv
source .venv/bin/activate
uv pip install fastapi uvicorn amplifier-core amplifier-foundation

# Run backend
python main.py

# Run frontend (separate terminal)
cd frontend
npm install
npm run dev

Key Design Decisions

What You Get

• ✅ One Amplifier session per email (isolated context)
• ✅ Deterministic session IDs (reproducible)
• ✅ Auto-refresh every minute (background task)
• ✅ Real-time streaming (SSE)
• ✅ Custom tools (reply, archive)
• ✅ Approval gates (safety)
• ✅ Web UI (accessible from anywhere)

Extensions

Easy additions to this foundation:

• Calendar integration - Extract dates, create events
• Multi-account - Handle multiple email addresses
• Smart labels - Auto-categorize emails
• Templates - Store common reply patterns
• Slack integration - Notify team of important emails
• Analytics - Track response times, AI accuracy

General Patterns for Building Apps

What to Reuse from Amplifier

• amplifier-core - Session orchestration (mandatory)
• amplifier-foundation - Bundle loading, module resolution (highly recommended)
• Provider modules - Anthropic/OpenAI/etc. integrations (save months of work)
• Tool modules - Filesystem, bash, web (where appropriate)
• Hook modules - Logging, redaction, streaming (adapt to your UI)
• Bundles - Use existing or create custom

What to Build Custom

• UI layer - Your interface (web/mobile/desktop)
• Session storage - CLI uses filesystem, you might use database/cloud
• Settings management - CLI uses YAML, you might use UI/API
• Domain tools - Email operations, calendar, CRM, etc.
• Authentication - User accounts, permissions

Additional Resources

• amplifier-core/examples/ - Reference implementations
• amplifier-foundation/examples/ - Bundle usage examples
• amplifier-app-transcribe - Community app example
• amplifier-app-blog-creator - Another community app