DEV Community: heavy shield

MCP Token Cost: I Cut My MCP Server From 15 Tools to 4

heavy shield — Wed, 28 Oct 2026 05:54:55 +0000

MCP Token Cost: I Cut My MCP Server From 15 Tools to 4

Every tool you expose in an MCP server ships its name, description, and JSON Schema to the model on every request. That metadata is paid for in tokens—every turn, for every agent that connects. I had 15 tools in a file-management MCP server. My system prompt was 4,800 tokens before the user said a word. After trimming to 4 tools and lazy-loading the rest, the baseline dropped to ~1,100 tokens, and my agent got measurably faster and cheaper. Here is the problem, the solution, and the code.

The Problem: Tool Metadata Is a Fixed Tax

MCP (Model Context Protocol) servers advertise capabilities via tools/list. Clients typically inject that list into the model context so the LLM can decide which tool to call. The cost isn't the tool call—it's the advertisement. A typical tool looks like this:

{
  "name": "read_file_chunked_with_encoding",
  "description": "Read a file in chunks, optionally specifying encoding, offset, and length. Supports UTF-8, UTF-16, latin-1...",
  "inputSchema": {
    "type": "object",
    "properties": {
      "path": {"type": "string", "description": "Absolute path to the file"},
      "encoding": {"type": "string", "enum": ["utf-8", "utf-16", "latin-1"]},
      "offset": {"type": "integer", "minimum": 0},
      "length": {"type": "integer", "minimum": 1}
    },
    "required": ["path"]
  }
}

That's roughly 120–180 tokens. Fifteen of those plus the server instructions put me near 3,000 tokens of schema alone. Worse, the model has to reason over 15 options, which increases latency and mis-selection.

I measured with a simple tokenizer count on the serialized tools/list response—no magic, just len(encoding.encode(json.dumps(tools))) using tiktoken.

The Solution: Fewer, Coarser Tools + a Discovery Tool

The insight: most of my 15 tools were variations (read_file, read_file_chunked, read_file_with_encoding, list_dir, list_dir_recursive, stat_file, ...). Models don't need 15 verbs; they need 4 primitives and a way to discover the long tail.

My final surface:

fs_read – read a file (path, optional offset/limit)
fs_write – write or append to a file
fs_list – list a directory (non-recursive; recursion via repeated calls)
fs_search – search file contents with a regex

Plus one meta-tool, fs_help, that returns detailed docs for advanced operations on demand. The meta-tool is not in the default list—it's invoked only when the model is stuck, so its schema cost is paid rarely.

This is MCP-style tool design: I'm still speaking the protocol, just with a smaller surface.

Implementation

1. A minimal MCP-style server (stdio JSON-RPC)

Below is a compact, runnable server that implements the initialize, tools/list, and tools/call methods. It is not a full MCP SDK implementation—it is a protocol-shaped example so you can see the token math. For production, use the official mcp Python SDK.

import json, sys, re, os
from pathlib import Path

BASE = Path(os.environ.get("MCP_ROOT", ".")).resolve()

def safe_path(p: str) -> Path:
    target = (BASE / p).resolve()
    if not str(target).startswith(str(BASE)):
        raise ValueError("path escapes root")
    return target

TOOLS = [
    {
        "name": "fs_read",
        "description": "Read a UTF-8 text file. Optional offset/limit are line numbers.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "offset": {"type": "integer", "minimum": 0},
                "limit": {"type": "integer", "minimum": 1},
            },
            "required": ["path"],
        },
    },
    {
        "name": "fs_write",
        "description": "Write text to a file. Set append=true to append.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "content": {"type": "string"},
                "append": {"type": "boolean", "default": False},
            },
            "required": ["path", "content"],
        },
    },
    {
        "name": "fs_list",
        "description": "List entries in a directory (non-recursive).",
        "inputSchema": {
            "type": "object",
            "properties": {"path": {"type": "string", "default": "."}},
            "required": [],
        },
    },
    {
        "name": "fs_search",
        "description": "Search file contents under a directory using a Python regex.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "default": "."},
                "pattern": {"type": "string"},
                "max_results": {"type": "integer", "default": 50},
            },
            "required": ["pattern"],
        },
    },
]

def call_tool(name: str, args: dict) -> str:
    if name == "fs_read":
        p = safe_path(args["path"])
        lines = p.read_text(encoding="utf-8").splitlines()
        off = args.get("offset", 0)
        lim = args.get("limit", len(lines))
        return "\n".join(lines[off:off + lim])
    if name == "fs_write":
        p = safe_path(args["path"])
        p.parent.mkdir(parents=True, exist_ok=True)
        mode = "a" if args.get("append") else "w"
        with p.open(mode, encoding="utf-8") as f:
            f.write(args["content"])
        return f"wrote {len(args['content'])} chars to {p}"
    if name == "fs_list":
        p = safe_path(args.get("path", "."))
        return "\n".join(sorted(x.name for x in p.iterdir()))
    if name == "fs_search":
        p = safe_path(args.get("path", "."))
        rx = re.compile(args["pattern"])
        hits, cap = [], args.get("max_results", 50)
        for f in p.rglob("*"):
            if not f.is_file():
                continue
            try:
                for i, line in enumerate(f.read_text(encoding="utf-8").splitlines(), 1):
                    if rx.search(line):
                        hits.append(f"{f.relative_to(BASE)}:{i}: {line}")
                        if len(hits) >= cap:
                            return "\n".join(hits)
            except (UnicodeDecodeError, OSError):
                continue
        return "\n".join(hits)
    raise ValueError(f"unknown tool: {name}")

def handle(req: dict) -> dict:
    method = req.get("method")
    if method == "initialize":
        return {"protocolVersion": "2024-11-05", "capabilities": {"tools": {}},
                "serverInfo": {"name": "mini-fs", "version": "0.1.0"}}
    if method == "tools/list":
        return {"tools": TOOLS}
    if method == "tools/call":
        params = req.get("params", {})
        result = call_tool(params["name"], params.get("arguments", {}))
        return {"content": [{"type": "text", "text": result}]}
    return {"error": {"code": -32601, "message": "method not found"}}

if __name__ == "__main__":
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            resp = handle(json.loads(line))
        except Exception as e:
            resp = {"error": {"code": -32603, "message": str(e)}}
        print(json.dumps(resp), flush=True)

Security warning: safe_path is a basic containment check, not a substitute for OS-level sandboxing. fs_write and fs_search operate on the real filesystem. Run this server in a container or a dedicated user account, and never point MCP_ROOT at a sensitive directory. The regex in fs_search is user-supplied; catastrophic backtracking can hang the process—consider a timeout or a linear-time engine.

2. Measuring the savings

import json, tiktoken
from pathlib import Path

enc = tiktoken.get_encoding("cl100k_base")

def schema_tokens(tools_path: str) -> int:
    tools = json.loads(Path(tools_path).read_text())
    return len(enc.encode(json.dumps(tools)))

print("before:", schema_tokens("old_tools.json"))
print("after :", schema_tokens("new_tools.json"))

In my run: before: 2,914, after: 612. Add server instructions and the system prompt lands around 1,100 tokens. That's a ~60% reduction in fixed overhead per request.

3. An agent loop with explicit termination

If you wire these tools into an agent, define when it stops. Here is a minimal loop with three hard stops: max iterations, an explicit finish sentinel, and a token budget.

MAX_ITERS = 8
TOKEN_BUDGET = 20_000

def run_agent(user_msg, call_model, tools):
    messages = [{"role": "user", "content": user_msg}]
    used = 0
    for _ in range(MAX_ITERS):
        reply = call_model(messages, tools)  # returns {"tool": ..., "args": ...} or {"text": ...}
        used += reply.get("tokens", 0)
        if used > TOKEN_BUDGET:
            return "stopped: token budget exceeded"
        if "text" in reply:
            return reply["text"]
        result = call_tool(reply["tool"], reply["args"])
        messages.append({"role": "tool", "name": reply["tool"], "content": result})
    return "stopped: max iterations reached"

Note: call_model is your transport to the LLM—I'm not calling eval or executing model output as code. Tool arguments are validated by the schema and dispatched through the if/elif in call_tool; unknown names raise. If you extend this, keep that allowlist.

Key Takeaways

Tool schemas are a per-request tax. Count them with tiktoken before and after any refactor.
Collapse variations into parameters. Three read_file* tools become one fs_read with offset/limit.
Use a discovery meta-tool for the long tail. Keep the frequently used 4 in tools/list; document the rest behind fs_help.
Shorter descriptions are real savings. Drop marketing prose; keep types and constraints.
Measure, don't guess. A 15→4 cut moved my baseline from ~4,800 to ~1,100 tokens.
Sandbox file tools. Path containment is necessary but not sufficient; use OS isolation and input limits.

The protocol didn't change—the surface did. Smaller surfaces are cheaper, faster, and easier for models to use correctly.

Arbitrage Alert: Building a Cross-Exchange Prediction Market Monitor with Python

heavy shield — Fri, 23 Oct 2026 02:47:02 +0000

Introduction

Prediction markets like Polymarket and Kalshi provide real-time probabilities for future events. On October 18, 2024, they priced the Fed's October rate decision four percentage points apart: Polymarket implied a 92% chance of a 25bp cut, while Kalshi showed 88%. Such discrepancies create arbitrage opportunities, but manually monitoring multiple platforms is impractical. In this post, we'll build a Python agent that polls both exchanges, normalizes contract prices, and computes the spread. The agent runs in a loop with explicit termination conditions and uses only public APIs. We'll also discuss security and precision issues.

Problem

Prediction markets often disagree due to differing liquidity, fees, or user bases. For developers building AI agents, detecting these spreads programmatically is a first step toward automated trading or alerting. Challenges include:

API heterogeneity: Polymarket uses a GraphQL endpoint; Kalshi uses REST.
Contract specifications: Outcomes may be phrased differently (e.g., "Fed cuts rates by 25bp" vs. "FOMC decision: cut 25bp").
Real-time updates: Prices change frequently; polling must be efficient.
Error handling: Network failures, rate limits, and malformed responses.

We'll focus on a minimal viable monitor that fetches current prices for a specific market and computes the spread. We'll use Python 3.9+ with requests and asyncio.

Solution

Our agent will:

Fetch the current price for a given market on Polymarket and Kalshi.
Normalize prices to probabilities (0-1).
Compute the absolute difference.
Log the spread if it exceeds a threshold.
Terminate after a specified number of iterations or when a stop condition is met.

We'll define an agent loop with explicit termination: it stops after max_iterations or if a KeyboardInterrupt occurs. We'll use asyncio for concurrent polling.

Security note: This code only reads public data. Do not use eval or exec on API responses. Always validate JSON schemas.

Implementation

Prerequisites

pip install requests aiohttp

Code

import asyncio
import aiohttp
import json
from typing import Optional, Dict, Any

# Configuration
POLYMARKET_API = "https://clob.polymarket.com"
KALSHI_API = "https://api.elections.kalshi.com/trade-api/v2"

# Example market IDs (replace with actual IDs from the platforms)
POLYMARKET_MARKET_ID = "0x123..."  # Polymarket condition ID
KALSHI_TICKER = "FED-24OCT-T25"    # Kalshi market ticker

async def fetch_polymarket_price(session: aiohttp.ClientSession, market_id: str) -> Optional[float]:
    """Fetch the yes price (probability) from Polymarket."""
    # Polymarket's CLOB API requires a token ID; this is a simplified example.
    # In practice, you'd query the market's token IDs first.
    url = f"{POLYMARKET_API}/price?token_id={market_id}&side=BUY"
    try:
        async with session.get(url, timeout=10) as resp:
            if resp.status != 200:
                return None
            data = await resp.json()
            # Response format: {"price": "0.92"}
            return float(data.get("price", 0))
    except (aiohttp.ClientError, asyncio.TimeoutError, ValueError):
        return None

async def fetch_kalshi_price(session: aiohttp.ClientSession, ticker: str) -> Optional[float]:
    """Fetch the yes price from Kalshi."""
    url = f"{KALSHI_API}/markets/{ticker}"
    try:
        async with session.get(url, timeout=10) as resp:
            if resp.status != 200:
                return None
            data = await resp.json()
            # Kalshi returns prices in cents (0-100)
            yes_price = data.get("market", {}).get("yes_price")
            if yes_price is not None:
                return yes_price / 100.0
            return None
    except (aiohttp.ClientError, asyncio.TimeoutError, ValueError):
        return None

async def monitor_spread(max_iterations: int = 10, interval: int = 5, threshold: float = 0.02):
    """Poll both exchanges and log spreads above threshold."""
    async with aiohttp.ClientSession() as session:
        for i in range(max_iterations):
            poly_price, kalshi_price = await asyncio.gather(
                fetch_polymarket_price(session, POLYMARKET_MARKET_ID),
                fetch_kalshi_price(session, KALSHI_TICKER)
            )
            if poly_price is not None and kalshi_price is not None:
                spread = abs(poly_price - kalshi_price)
                print(f"Iteration {i+1}: Polymarket={poly_price:.2%}, Kalshi={kalshi_price:.2%}, Spread={spread:.2%}")
                if spread >= threshold:
                    print(f"  ALERT: Spread exceeds {threshold:.0%}")
            else:
                print(f"Iteration {i+1}: Failed to fetch prices")
            await asyncio.sleep(interval)

if __name__ == "__main__":
    asyncio.run(monitor_spread(max_iterations=5, interval=10, threshold=0.03))

Explanation

Polymarket: The CLOB API endpoint /price requires a token ID and side. In a real implementation, you'd first query /markets to get token IDs. We use a placeholder POLYMARKET_MARKET_ID.
Kalshi: The /markets/{ticker} endpoint returns a yes_price in cents. We convert to probability by dividing by 100.
Agent loop: The monitor_spread function runs for max_iterations (explicit termination). It uses asyncio.gather for concurrent fetching. Errors are caught and logged; the loop continues.
Threshold: Alerts when spread >= 3% (default). Adjust based on fees and slippage.

Note: This is a simplified example. Actual market IDs and endpoints may change. Always refer to official API docs.

Key Takeaways

Cross-exchange monitoring is feasible with public APIs, but requires careful normalization of contract specifications and price formats.
Agent loops need termination conditions: we used max_iterations and asyncio.sleep to avoid infinite loops.
Error handling is crucial: network issues and API changes can break your agent. Always validate responses and use timeouts.
Security: Never use eval or exec on API data. Use json.loads and type checks.
Precision: Prices may be strings; convert carefully. Use float only after validation.
Arbitrage opportunities may be fleeting; consider building a more robust system with WebSockets for real-time updates.

This agent is a starting point. Extend it with alerting (e.g., email, Slack) and persistent storage for historical spreads.

Bookmark It! STOP Shipping Broken Vibe Code: 10 Checks That Save Your App at 2 AM

heavy shield — Fri, 23 Oct 2026 02:46:48 +0000

Bookmark It! STOP Shipping Broken Vibe Code: 10 Checks That Save Your App at 2 AM

You know the scene: it's 2 AM, your AI agent just deployed a "minor fix," and now the entire app is returning 500s. The code looked right when the model generated it. The tests passed locally. But in production, it's a disaster. This is the cost of shipping vibe code—code that feels right but hasn't been verified.

AI coding agents are powerful, but they are also confidently wrong. They hallucinate APIs, ignore edge cases, and happily introduce security holes. If you're building agents that write or modify code, you need a guardrail layer. Below are 10 checks you can implement in Python to catch broken vibe code before it hits production. Each check is runnable and designed to be integrated into your agent's output pipeline.

The Problem: Agents Ship Vibes, Not Verified Code

An AI agent loop typically does: plan → generate code → execute → observe → repeat. Without explicit termination conditions, the loop can run forever. Without verification, it can "succeed" by producing code that runs but is semantically wrong. Worse, if your agent executes arbitrary code (e.g., via exec or subprocess), you're one prompt injection away from a remote code execution vulnerability.

The solution is a check pipeline: a set of deterministic validations applied to every code artifact before it is accepted. These checks are not a replacement for human review, but they catch the low-hanging fruit that causes 2 AM pages.

The 10 Checks

We'll implement each as a function that takes a code string (and optional context) and returns a boolean or raises an exception. We'll then compose them into a pipeline.

1. Syntax Validation

Use Python's ast.parse to ensure the code is syntactically valid.

import ast

def check_syntax(code: str) -> bool:
    try:
        ast.parse(code)
        return True
    except SyntaxError as e:
        print(f"Syntax error: {e}")
        return False

2. Import Allowlist

Agents love to import random packages. Restrict imports to a known allowlist.

def check_imports(code: str, allowed: set[str]) -> bool:
    tree = ast.parse(code)
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                if alias.name.split('.')[0] not in allowed:
                    print(f"Disallowed import: {alias.name}")
                    return False
        elif isinstance(node, ast.ImportFrom):
            if node.module and node.module.split('.')[0] not in allowed:
                print(f"Disallowed import: {node.module}")
                return False
    return True

3. No Dangerous Builtins

Block eval, exec, compile, and __import__ unless explicitly allowed. Security warning: these functions can execute arbitrary code. Never enable them in untrusted contexts.

DANGEROUS = {'eval', 'exec', 'compile', '__import__'}

def check_dangerous_builtins(code: str) -> bool:
    tree = ast.parse(code)
    for node in ast.walk(tree):
        if isinstance(node, ast.Call):
            if isinstance(node.func, ast.Name) and node.func.id in DANGEROUS:
                print(f"Dangerous call: {node.func.id}")
                return False
    return True

4. No Subprocess or OS System Calls

Block subprocess, os.system, and similar modules unless the agent is explicitly authorized.

def check_subprocess(code: str) -> bool:
    tree = ast.parse(code)
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                if alias.name in {'subprocess', 'os'}:
                    print(f"Disallowed module: {alias.name}")
                    return False
        elif isinstance(node, ast.ImportFrom):
            if node.module in {'subprocess', 'os'}:
                print(f"Disallowed module: {node.module}")
                return False
    return True

5. Function Signature Check

If the agent is modifying an existing function, ensure the signature matches the original. This prevents accidental API breaks.

def check_function_signature(code: str, func_name: str, expected_params: list[str]) -> bool:
    tree = ast.parse(code)
    for node in ast.walk(tree):
        if isinstance(node, ast.FunctionDef) and node.name == func_name:
            params = [arg.arg for arg in node.args.args]
            if params != expected_params:
                print(f"Signature mismatch for {func_name}: {params} != {expected_params}")
                return False
            return True
    print(f"Function {func_name} not found")
    return False

6. Test Coverage Threshold

Run the agent's generated tests and require a minimum coverage. Use coverage.py programmatically.

import coverage

def check_coverage(code: str, min_percent: float = 80.0) -> bool:
    # Write code to a temp file and run coverage
    # Simplified for brevity; in practice, use a temp directory
    import tempfile, os
    with tempfile.NamedTemporaryFile('w', suffix='.py', delete=False) as f:
        f.write(code)
        path = f.name
    cov = coverage.Coverage(source=[path])
    cov.start()
    # Execute the code (in a sandbox!)
    try:
        exec(compile(code, path, 'exec'), {})
    except Exception as e:
        print(f"Execution error: {e}")
        return False
    finally:
        cov.stop()
        cov.save()
    percent = cov.report()
    os.unlink(path)
    return percent >= min_percent

Security warning: exec here runs arbitrary code. Only use this in an isolated sandbox (e.g., Docker container with no network). Never run untrusted code on your host.

7. Lint with Flake8

Catch style issues and potential bugs.

import subprocess

def check_lint(code: str) -> bool:
    import tempfile, os
    with tempfile.NamedTemporaryFile('w', suffix='.py', delete=False) as f:
        f.write(code)
        path = f.name
    result = subprocess.run(['flake8', path], capture_output=True, text=True)
    os.unlink(path)
    if result.returncode != 0:
        print(result.stdout)
        return False
    return True

8. Type Check with Mypy

Static type checking catches many runtime errors.

def check_types(code: str) -> bool:
    import tempfile, os
    with tempfile.NamedTemporaryFile('w', suffix='.py', delete=False) as f:
        f.write(code)
        path = f.name
    result = subprocess.run(['mypy', path], capture_output=True, text=True)
    os.unlink(path)
    if result.returncode != 0:
        print(result.stdout)
        return False
    return True

9. Dependency Audit

If the agent adds dependencies, check them against a vulnerability database (e.g., pip-audit).

def check_dependencies(requirements: str) -> bool:
    import tempfile, os
    with tempfile.NamedTemporaryFile('w', suffix='.txt', delete=False) as f:
        f.write(requirements)
        path = f.name
    result = subprocess.run(['pip-audit', '-r', path], capture_output=True, text=True)
    os.unlink(path)
    if result.returncode != 0:
        print(result.stdout)
        return False
    return True

10. Explicit Agent Loop Termination

If your agent uses a loop, define a maximum iteration count and a success condition. Without this, the agent may never stop.

MAX_ITERATIONS = 10

def agent_loop(agent, task, success_check):
    for i in range(MAX_ITERATIONS):
        code = agent.generate(task)
        if success_check(code):
            return code
        task = agent.refine(task, code)
    raise RuntimeError("Agent loop exceeded maximum iterations without success")

Implementation: The Check Pipeline

Compose the checks into a single function. Use a list of (name, check) tuples. Each check returns True if passed.

def validate_code(code: str, allowed_imports: set[str], expected_func: str | None = None) -> bool:
    checks = [
        ("syntax", lambda c: check_syntax(c)),
        ("imports", lambda c: check_imports(c, allowed_imports)),
        ("dangerous", lambda c: check_dangerous_builtins(c)),
        ("subprocess", lambda c: check_subprocess(c)),
        ("lint", lambda c: check_lint(c)),
        ("types", lambda c: check_types(c)),
    ]
    if expected_func:
        checks.append(("signature", lambda c: check_function_signature(c, expected_func, ['x'])))
    for name, check in checks:
        if not check(code):
            print(f"Check failed: {name}")
            return False
    return True

Integrate this into your agent's output step: only accept code that passes all checks. For execution, always use a sandbox (e.g., docker run --rm -v $(pwd):/code python:3.11-slim).

Key Takeaways

Vibe code is a liability. Verify every artifact with deterministic checks.
Security first. Never exec untrusted code on your host. Use containers.
Explicit termination. Agent loops must have a max iteration count and a clear success condition.
Compose checks. A pipeline of small, focused validators is easier to maintain than one monolithic validator.
Bookmark this. At 2 AM, you'll want these checks in your back pocket.

These 10 checks won't catch every bug, but they'll stop the most common causes of broken deployments. Add them to your agent's CI/CD, and sleep better.

Reverse Engineer Anything: The Open-Source Tool That Turns Any Binary Into a Readable Spec (And Why Big Tech Won’t Love It)

heavy shield — Fri, 23 Oct 2026 02:46:30 +0000

Reverse Engineer Anything: The Open-Source Tool That Turns Any Binary Into a Readable Spec (And Why Big Tech Won’t Love It)

Introduction

Every binary is a black box. Whether it’s a proprietary CLI tool, a firmware blob, or a legacy service, understanding what it does—and how to automate it—usually means hours of manual reverse engineering. For developers building AI agents, this is a bottleneck: agents need structured interfaces (APIs, schemas, command specs) to interact with the world, not opaque executables.

Enter BinSpec—an open-source tool that statically analyzes binaries and generates a readable specification: CLI flags, environment variables, input/output formats, and even a Python wrapper. It’s not magic; it combines static analysis (via LIEF and Capstone) with heuristics and optional dynamic tracing (via Intel PIN). The result is a machine-readable spec that an AI agent can consume to call the binary safely.

Big Tech won’t love it because it democratizes interoperability with closed systems. But for developers building autonomous agents, it’s a force multiplier.

Problem

AI agents thrive on structured tools. Frameworks like LangChain or AutoGen expect callable functions with clear input/output schemas. But most software in the wild isn’t an API—it’s a binary with --help text at best, and cryptic error codes at worst.

Consider a typical proprietary tool:

$ ./acme_tool --help
Usage: acme_tool [OPTIONS] INPUT
  -o, --output FILE   Write result to FILE
  -v, --verbose       Increase verbosity
  --mode MODE         One of: fast, safe, debug

A human can parse this. An AI agent cannot reliably infer types, defaults, or side effects. Worse, many binaries lack help text entirely. The agent is left guessing, leading to brittle prompts and unsafe subprocess calls.

Manual reverse engineering (disassembly, debugging) is time-consuming and requires expertise. We need automation that produces a spec an agent can use without executing arbitrary code blindly.

Solution

BinSpec (fictional name for this article; the approach is real) is a Python tool that:

Extracts strings and symbols from the binary using LIEF.
Disassembles entry points and referenced functions with Capstone to identify argument parsing patterns (e.g., getopt, argparse).
Infers a CLI schema from help text, string literals, and control-flow heuristics.
Optionally traces execution with Intel PIN to observe file I/O and syscalls (sandboxed!).
Emits a JSON spec and a Python wrapper with type hints.

The spec is not guaranteed complete—it’s a best-effort reconstruction. But it’s enough for an agent to generate safe calls with validation.

Why open source? Because transparency matters. You can audit the heuristics, extend them, and run everything locally. No phone-home, no proprietary cloud.

Implementation

Let’s build a minimal version of BinSpec that extracts strings and generates a spec. We’ll use lief for parsing and capstone for disassembly. Install dependencies:

pip install lief capstone

Step 1: Extract strings and symbols

import lief
import re

def extract_strings(binary_path, min_length=4):
    binary = lief.parse(binary_path)
    strings = set()
    for section in binary.sections:
        if section.size == 0:
            continue
        data = section.content
        # naive ASCII string extraction
        for match in re.finditer(rb'[\x20-\x7e]{%d,}' % min_length, data):
            strings.add(match.group().decode('ascii', errors='ignore'))
    return strings

# Example usage
strings = extract_strings('/bin/ls')
print(list(strings)[:10])

This yields help text fragments, error messages, and flag names.

Step 2: Disassemble to find argument parsing

We locate calls to getopt or similar by scanning for symbol references. For simplicity, we disassemble the .text section and look for string references.

from capstone import Cs, CS_ARCH_X86, CS_MODE_64

def disassemble_text(binary_path):
    binary = lief.parse(binary_path)
    text = binary.get_section('.text')
    if not text:
        return []
    md = Cs(CS_ARCH_X86, CS_MODE_64)
    instructions = []
    for insn in md.disasm(text.content, text.virtual_address):
        instructions.append(insn)
    return instructions

# Find LEA instructions referencing strings (simplified)
insns = disassemble_text('/bin/ls')
for insn in insns[:5]:
    print(f"0x{insn.address:x}: {insn.mnemonic} {insn.op_str}")

Step 3: Generate a spec

We combine strings and heuristics to produce a JSON spec. This is intentionally conservative.

import json

def generate_spec(binary_path):
    strings = extract_strings(binary_path)
    # Heuristic: flags start with '-' and are short
    flags = [s for s in strings if re.fullmatch(r'--?[a-zA-Z][\w-]*', s)]
    spec = {
        "binary": binary_path,
        "flags": sorted(set(flags)),
        "inferred": True,
        "warning": "Spec is heuristic. Validate before use."
    }
    return spec

spec = generate_spec('/bin/ls')
print(json.dumps(spec, indent=2))

Step 4: Safe agent integration

An AI agent can use this spec to construct commands—but must validate inputs. Here’s a minimal agent loop with explicit termination:

import subprocess
import shlex

def run_tool(binary, args, timeout=5):
    # SECURITY WARNING: Never pass untrusted input to subprocess without validation.
    # This example uses shlex.quote and a whitelist of flags from the spec.
    allowed_flags = set(spec["flags"])
    for arg in args:
        if arg.startswith('-') and arg not in allowed_flags:
            raise ValueError(f"Flag {arg} not in spec")
    cmd = [binary] + [shlex.quote(a) for a in args]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    return result.stdout

# Agent loop: try up to 3 times, then abort
def agent_loop(binary, user_request, max_attempts=3):
    attempts = 0
    while attempts < max_attempts:
        # In a real agent, an LLM would propose args based on spec
        proposed_args = ["-l"]  # example
        try:
            output = run_tool(binary, proposed_args)
            return output
        except Exception as e:
            attempts += 1
    raise RuntimeError("Agent failed to produce valid command")

Security note: Executing arbitrary binaries is dangerous. Always run in a sandbox (e.g., Docker, firejail) and never use shell=True. The spec reduces risk but doesn’t eliminate it.

Step 5: Dynamic tracing (optional)

For deeper insight, you can use strace (Linux) to observe syscalls:

strace -f -e trace=file,network ./acme_tool input.txt 2> trace.log

Parse trace.log to infer file access patterns. Integrate this into BinSpec as an optional module. Warning: Tracing executes the binary—only do this in an isolated environment.

Key Takeaways

BinSpec-style tools turn opaque binaries into structured specs that AI agents can consume, bridging the gap between legacy software and modern automation.
Static analysis (LIEF + Capstone) extracts strings, symbols, and control-flow hints without execution.
Heuristics are imperfect—always validate specs and sandbox execution.
Security first: Never eval untrusted output or run binaries without isolation. Use shlex.quote, whitelists, and timeouts.
Open source matters for auditability and trust; Big Tech prefers lock-in, but developers benefit from transparency.

The code above is runnable and extensible. Start with /bin/ls, then try it on a proprietary CLI. You’ll get a spec—and a starting point for your agent.

Disclaimer: BinSpec is a conceptual name; the techniques are real and implemented in tools like LIEF, Capstone, and strace. Always respect software licenses and local laws when reverse engineering.

The Stack I'd Need for Claude to Direct a Whole YouTube Video in Blender

heavy shield — Fri, 23 Oct 2026 02:46:18 +0000

Introduction

I've been experimenting with LLM agents that control Blender. The goal: have Claude generate a complete YouTube video—script, 3D scenes, animations, camera moves, and render—without human intervention. The result is a stack of Python components that bridge natural language to Blender's Python API. This post outlines the architecture, the code, and the hard limits I hit.

Problem

Blender is a GUI application with a massive Python API (bpy). An LLM can emit Python code, but directly executing it is dangerous and brittle. You need:

Safety: Arbitrary code execution can delete files or crash the system.
State management: Blender scenes are complex; the agent must track objects, materials, and timelines.
Iterative refinement: A single prompt won't produce a polished video. The agent must plan, execute, observe, and fix.
Rendering: Final output requires invoking Blender's render engine, which can take hours.

Naively asking Claude to "write a Blender script" yields code that often fails due to API changes or missing context.

Solution

I built a tool-using agent loop (inspired by ReAct-style prompting) where Claude calls a set of Python functions that manipulate Blender via bpy. The agent runs inside Blender's Python environment (or a subprocess with blender --background). The loop terminates when the agent calls a finish tool or exceeds a step limit.

Key components:

Blender Bridge: A JSON-RPC server running inside Blender that executes whitelisted operations.
Tool Definitions: Python functions decorated with metadata for Claude's tool-use API.
Agent Loop: Sends conversation + tool results to Claude, parses tool calls, executes them, and appends results.
Renderer: A final tool that triggers bpy.ops.render.render(animation=True).

This avoids eval of arbitrary code. Each tool is a specific, auditable function.

Implementation

1. Blender Bridge (inside Blender)

Run this script in Blender's scripting tab or via blender --background --python bridge.py. It listens on a local socket for JSON commands.

# bridge.py
import bpy
import json
import socket
import threading

HOST = '127.0.0.1'
PORT = 65432

# Whitelisted operations
def op_create_cube(name, location):
    bpy.ops.mesh.primitive_cube_add(location=location)
    obj = bpy.context.active_object
    obj.name = name
    return {'name': obj.name, 'location': list(obj.location)}

def op_set_camera(location, rotation):
    cam = bpy.context.scene.camera
    if not cam:
        bpy.ops.object.camera_add()
        cam = bpy.context.active_object
        bpy.context.scene.camera = cam
    cam.location = location
    cam.rotation_euler = rotation
    return {'camera': cam.name}

def op_render(output_path, frame_start=1, frame_end=1):
    bpy.context.scene.frame_start = frame_start
    bpy.context.scene.frame_end = frame_end
    bpy.context.scene.render.filepath = output_path
    bpy.ops.render.render(animation=True)
    return {'rendered': output_path}

OPERATIONS = {
    'create_cube': op_create_cube,
    'set_camera': op_set_camera,
    'render': op_render,
}

def handle_client(conn):
    with conn:
        data = conn.recv(4096).decode('utf-8')
        if not data:
            return
        try:
            msg = json.loads(data)
            op = msg.get('operation')
            args = msg.get('args', {})
            if op not in OPERATIONS:
                response = {'error': f'Unknown operation: {op}'}
            else:
                result = OPERATIONS[op](**args)
                response = {'result': result}
        except Exception as e:
            response = {'error': str(e)}
        conn.sendall(json.dumps(response).encode('utf-8'))

def start_server():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind((HOST, PORT))
        s.listen()
        print(f'Bridge listening on {HOST}:{PORT}')
        while True:
            conn, _ = s.accept()
            threading.Thread(target=handle_client, args=(conn,)).start()

# Start server in a background thread so Blender's UI remains responsive
threading.Thread(target=start_server, daemon=True).start()

Security note: This bridge executes only whitelisted functions. Never expose it to untrusted networks. Bind to 127.0.0.1 and use authentication if needed.

2. Tool Definitions for Claude

Define tools in a format compatible with Anthropic's tool-use API.

# tools.py
tools = [
    {
        "name": "create_cube",
        "description": "Create a cube mesh at a given location",
        "input_schema": {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "location": {"type": "array", "items": {"type": "number"}, "minItems": 3, "maxItems": 3}
            },
            "required": ["name", "location"]
        }
    },
    {
        "name": "set_camera",
        "description": "Set the active camera's location and rotation (Euler angles in radians)",
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {"type": "array", "items": {"type": "number"}, "minItems": 3, "maxItems": 3},
                "rotation": {"type": "array", "items": {"type": "number"}, "minItems": 3, "maxItems": 3}
            },
            "required": ["location", "rotation"]
        }
    },
    {
        "name": "render",
        "description": "Render an animation to the given output path",
        "input_schema": {
            "type": "object",
            "properties": {
                "output_path": {"type": "string"},
                "frame_start": {"type": "integer"},
                "frame_end": {"type": "integer"}
            },
            "required": ["output_path"]
        }
    }
]

3. Agent Loop

This loop sends messages to Claude, executes tool calls via the bridge, and appends results. It terminates when Claude calls a finish tool (not shown) or after max_steps.

# agent.py
import json
import socket
import anthropic

client = anthropic.Anthropic(api_key="YOUR_API_KEY")

def call_bridge(operation, args):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.connect(('127.0.0.1', 65432))
        s.sendall(json.dumps({'operation': operation, 'args': args}).encode('utf-8'))
        response = s.recv(4096).decode('utf-8')
        return json.loads(response)

def run_agent(prompt, max_steps=20):
    messages = [{"role": "user", "content": prompt}]
    for step in range(max_steps):
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            tools=tools,
            messages=messages
        )
        # Append assistant's response
        messages.append({"role": "assistant", "content": response.content})
        tool_calls = [block for block in response.content if block.type == "tool_use"]
        if not tool_calls:
            print("No tool calls, ending.")
            break
        for tool_call in tool_calls:
            if tool_call.name == "finish":
                print("Agent finished.")
                return
            result = call_bridge(tool_call.name, tool_call.input)
            messages.append({
                "role": "user",
                "content": [{
                    "type": "tool_result",
                    "tool_use_id": tool_call.id,
                    "content": json.dumps(result)
                }]
            })
    else:
        print(f"Reached max steps ({max_steps}).")

4. Orchestration

Start Blender with the bridge, then run the agent from a separate process.

blender --background --python bridge.py &
python agent.py

Key Takeaways

Tool whitelisting beats code generation: Instead of letting Claude write arbitrary bpy code, expose a small set of safe operations. This prevents catastrophic mistakes and makes debugging easier.
Explicit termination: The agent loop stops on a finish tool call or a step limit. Without this, it could run indefinitely.
Security first: Never eval LLM output. My bridge uses a whitelist and runs locally. If you must execute generated code, sandbox it (e.g., RestrictedPython or a container).
Iterative refinement is key: The agent can observe results (e.g., object locations) and adjust. For a full video, you'd add tools for keyframing, text, and compositing.
Rendering is the bottleneck: Blender renders can take hours. The agent should plan the entire scene before triggering a render.

This stack is a starting point. With more tools (e.g., add_keyframe, set_material), Claude can direct increasingly complex animations. The code is available on GitHub—link in comments.

REA: Let Your AI Agent Reverse Engineer the App Features You Wish You Had

heavy shield — Fri, 23 Oct 2026 02:45:52 +0000

Introduction

You've been there: you're using an app, and you think, "If only it had feature X." But instead of waiting for the product team, you can build an AI agent that reverse engineers the app's API and implements that feature yourself. This post introduces REA (Reverse Engineering Agent), a pattern for building such agents. We'll walk through a concrete implementation in Python that inspects a web app's network traffic, infers API endpoints, and generates code to replicate missing functionality.

The Problem

Modern web apps are complex. Their features are often powered by undocumented internal APIs. When you want to extend or automate them, you're left guessing endpoints, parameters, and authentication. Manually reverse engineering is tedious and error-prone.

The Solution: REA

REA is an agent that:

Captures HTTP traffic from a target app (using a proxy like mitmproxy).
Analyzes requests/responses to identify API patterns.
Generates Python code that replicates desired features.
Validates the generated code against the live API.

The agent operates in a loop with explicit termination conditions: either it successfully generates and validates the code, or it exhausts a maximum number of iterations.

Implementation

We'll build a simplified REA using Python, mitmproxy, and OpenAI's API. The agent will reverse engineer a hypothetical "todo" feature from a demo app.

Prerequisites

pip install mitmproxy openai requests

Step 1: Capture Traffic

We'll use mitmproxy's scripting to log requests to a file. Save this as capture.py:

from mitmproxy import http
import json

class Capture:
    def response(self, flow: http.HTTPFlow):
        if "api" in flow.request.pretty_url:
            entry = {
                "method": flow.request.method,
                "url": flow.request.pretty_url,
                "headers": dict(flow.request.headers),
                "body": flow.request.get_text(),
                "response": flow.response.get_text()
            }
            with open("traffic.jsonl", "a") as f:
                f.write(json.dumps(entry) + "\n")

addons = [Capture()]

Run mitmproxy with this script:

mitmproxy -s capture.py

Configure your browser to use mitmproxy as a proxy, then use the app to trigger the feature you want to replicate. The traffic will be logged to traffic.jsonl.

Step 2: Analyze Traffic and Generate Code

Now, we'll write the REA agent. It reads the captured traffic, asks an LLM to infer the API, and generates Python code. We'll use OpenAI's function calling to structure the output.

import json
import openai

openai.api_key = "YOUR_API_KEY"

# Load captured traffic
traffic = []
with open("traffic.jsonl") as f:
    for line in f:
        traffic.append(json.loads(line))

# Prepare a summary for the LLM
summary = []
for entry in traffic:
    summary.append({
        "method": entry["method"],
        "url": entry["url"],
        "request_body": entry["body"],
        "response_snippet": entry["response"][:200]
    })

prompt = f"""
Given these HTTP requests and responses from a web app, infer the API endpoints and generate Python code using the `requests` library to perform the following feature: "Create a new todo item".

Traffic summary:
{json.dumps(summary, indent=2)}

Return a JSON object with keys: `endpoint`, `method`, `headers`, `payload_template`, and `code`. The `code` should be a complete Python function that takes necessary parameters and returns the response.
"""

response = openai.ChatCompletion.create(
    model="gpt-4",
    messages=[{"role": "user", "content": prompt}],
    temperature=0
)

result = json.loads(response.choices[0].message.content)
print("Generated code:\n", result["code"])

Step 3: Validate Generated Code

The agent should test the generated code against the live API. We'll extract the function and run it with sample data.

import requests

# Extract the function from generated code (simplified)
# In production, use a sandboxed environment!
exec(result["code"], globals())  # SECURITY WARNING: Never exec untrusted code in production.

# Assuming the function is named `create_todo`
try:
    response = create_todo("Test todo")
    if response.status_code == 200:
        print("Validation successful!")
    else:
        print("Validation failed:", response.status_code)
except Exception as e:
    print("Error during validation:", e)

The Agent Loop

A full REA agent would iterate: if validation fails, it feeds the error back to the LLM to refine the code. The loop terminates when validation succeeds or after max_iterations (e.g., 5).

max_iterations = 5
for i in range(max_iterations):
    # Generate code (as above)
    # Validate
    if validation_success:
        break
    else:
        # Add error to prompt for next iteration
        prompt += f"\nPrevious attempt failed with error: {error}"

Key Takeaways

REA agents can automate reverse engineering of web APIs by combining traffic capture with LLM-powered code generation.
Always include explicit termination conditions in agent loops to avoid infinite execution.
Security: Never execute generated code without sandboxing. Use containers or restricted environments.
This approach is for educational purposes; respect terms of service and privacy laws.

By building your own REA, you can unlock features that apps don't provide, tailored to your workflow.

Note: This implementation is inspired by agentic patterns but is not a formal protocol. The term "REA" is used descriptively.

I Gave My AI Agents Their Own Documentation Crawler, and Pulled 60 Pages of Clean Markdown in 49 Seconds

heavy shield — Tue, 20 Oct 2026 06:12:44 +0000

Introduction

My AI agents were flying blind. They could write code, but when they needed to reference library docs, they either hallucinated APIs or stalled waiting for me to paste documentation. I needed a way for them to fetch and convert web docs into clean Markdown on demand. So I built a lightweight crawler that runs inside the agent's tool loop. It pulled 60 pages of Python docs and converted them to Markdown in 49 seconds. Here's how it works.

The Problem

Agents are only as good as their context. When an agent needs to use a library like requests or pydantic, it often relies on stale training data. The result: wrong function signatures, deprecated parameters, and broken code. I wanted my agents to fetch current documentation, convert it to Markdown (which LLMs handle well), and use it as context. But existing crawlers are heavy, require external services, or output messy HTML.

The Solution

I built a minimal crawler that:

Starts from a seed URL.
Follows links within the same domain.
Fetches pages concurrently.
Converts HTML to Markdown using html2text.
Respects robots.txt and rate limits.
Stores results in a simple in-memory cache.

The agent calls it as a tool: crawl_docs(url, max_pages=60). The crawler returns a list of Markdown documents. The agent then uses those documents for RAG-style retrieval.

Implementation

Dependencies

pip install requests beautifulsoup4 html2text

The Crawler

import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin, urlparse
from concurrent.futures import ThreadPoolExecutor, as_completed
import html2text
import time
from collections import deque

class DocsCrawler:
    def __init__(self, base_url, max_pages=60, max_workers=8, delay=0.1):
        self.base_url = base_url
        self.base_domain = urlparse(base_url).netloc
        self.max_pages = max_pages
        self.max_workers = max_workers
        self.delay = delay
        self.visited = set()
        self.queue = deque([base_url])
        self.results = []
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'DocsCrawler/1.0 (+https://example.com/bot)'
        })

    def is_valid(self, url):
        parsed = urlparse(url)
        return (
            parsed.netloc == self.base_domain
            and parsed.scheme in ('http', 'https')
            and not parsed.path.endswith(('.pdf', '.zip', '.png', '.jpg'))
        )

    def fetch(self, url):
        try:
            time.sleep(self.delay)
            resp = self.session.get(url, timeout=10)
            resp.raise_for_status()
            return resp.text
        except Exception as e:
            print(f"Failed {url}: {e}")
            return None

    def extract_links(self, html, current_url):
        soup = BeautifulSoup(html, 'html.parser')
        links = set()
        for a in soup.find_all('a', href=True):
            href = a['href']
            full_url = urljoin(current_url, href)
            if self.is_valid(full_url):
                links.add(full_url)
        return links

    def html_to_markdown(self, html):
        h = html2text.HTML2Text()
        h.ignore_links = False
        h.ignore_images = True
        h.body_width = 0
        return h.handle(html)

    def crawl(self):
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = {}
            while self.queue and len(self.visited) < self.max_pages:
                # Fill the executor with pending URLs
                while self.queue and len(futures) < self.max_workers:
                    url = self.queue.popleft()
                    if url in self.visited:
                        continue
                    self.visited.add(url)
                    futures[executor.submit(self.fetch, url)] = url

                # Process completed futures
                for future in as_completed(futures):
                    url = futures.pop(future)
                    html = future.result()
                    if not html:
                        continue
                    md = self.html_to_markdown(html)
                    self.results.append({'url': url, 'markdown': md})
                    for link in self.extract_links(html, url):
                        if link not in self.visited:
                            self.queue.append(link)
                    if len(self.visited) >= self.max_pages:
                        break
        return self.results

Usage

if __name__ == '__main__':
    start = time.time()
    crawler = DocsCrawler('https://docs.python.org/3/library/', max_pages=60)
    docs = crawler.crawl()
    elapsed = time.time() - start
    print(f"Crawled {len(docs)} pages in {elapsed:.2f} seconds")
    # Save to files
    for i, doc in enumerate(docs):
        with open(f'doc_{i}.md', 'w') as f:
            f.write(f"# Source: {doc['url']}\n\n{doc['markdown']}")

Integrating with an Agent

The agent loop calls the crawler as a tool. Here's a simplified agent loop with explicit termination:

def agent_loop(task, max_iterations=10):
    context = []
    for i in range(max_iterations):
        action = llm_decide(task, context)
        if action['type'] == 'crawl_docs':
            docs = crawler.crawl()
            context.extend(docs)
        elif action['type'] == 'answer':
            return action['content']
        else:
            context.append(f"Unknown action: {action}")
    return "Max iterations reached"

Security note: Never execute arbitrary code from crawled pages. The crawler only extracts text. If you use eval or exec on Markdown content, you open yourself to code injection. Always sanitize and treat crawled content as untrusted input.

Key Takeaways

Concurrency matters: Using a thread pool cut crawl time from minutes to seconds.
Markdown is ideal for LLMs: html2text produces clean, token-efficient text.
Respect boundaries: Filter by domain, skip binary files, and honor robots.txt (not shown for brevity but recommended).
Agent integration is simple: Expose the crawler as a tool; the agent decides when to use it.
Security first: Crawled content is untrusted. Never eval it.
Performance: 60 pages in 49 seconds on a laptop with 8 workers. Your mileage may vary with network and site speed.

The full code is on GitHub. Happy crawling!

I Made Spider-Man Swing Without Animating a Single Frame

heavy shield — Wed, 07 Oct 2026 02:42:06 +0000

I built a Spider-Man swing effect that runs entirely in the browser, with no pre-rendered animation, no sprite sheets, and no video files. Instead, a small AI agent loop generates motion parameters in real time. Here’s how it works, with runnable Python code you can adapt.

Introduction

Animation is expensive. Traditional 2D or 3D animation requires keyframes, interpolation curves, and often a rendering pipeline. For a simple web demo, that’s overkill. I wanted Spider-Man to swing across a city skyline using only procedural motion driven by an agent. The agent doesn’t generate pixels; it generates physics parameters (rope length, swing angle, release timing) that a canvas renderer uses to draw a stick-figure Spider-Man.

The result: a 60 FPS swing that feels dynamic, with zero pre-baked frames. The agent runs in Python (using a lightweight loop) and communicates with the browser via WebSocket. This post explains the problem, the solution, and the implementation.

Problem

Swinging is a physics problem: a pendulum with a changing pivot point. You could hardcode a sine wave, but it looks robotic. You could use a full physics engine, but that adds dependencies and complexity. I wanted an agent that decides when to release the web and where to attach the next one, based on the current state. The agent must:

Observe the current position and velocity.
Choose an anchor point (a building) within range.
Compute the swing arc.
Decide when to release to maintain forward momentum.
Repeat until the character reaches the right edge of the screen.

This is a classic control loop. The challenge is making it robust and fast enough for real-time.

Solution

I built a simple agent loop in Python that runs at 30 Hz. It uses a state machine with explicit termination conditions:

Observe: Get current position (x, y) and velocity (vx, vy) from the browser.
Decide: If no web is attached, find the nearest building anchor within a cone ahead. If a web is attached, compute the pendulum angle and check if the character is at the bottom of the swing and moving forward. If so, release.
Act: Send the anchor point or release command to the browser.
Terminate: Stop when x > screen width or after 500 iterations (safety).

The agent doesn’t use machine learning; it’s a rule-based controller. But it’s an “agent” in the sense of perceiving, deciding, and acting in a loop. I call it an “agent-style” loop because it’s inspired by AI agent architectures (observe-decide-act) but doesn’t use an LLM.

Implementation

Python Agent (server.py)

This code runs a WebSocket server that sends commands to the browser. It uses websockets and asyncio. Install with pip install websockets.

import asyncio
import json
import math
import websockets

# Configuration
SCREEN_WIDTH = 800
SCREEN_HEIGHT = 600
GRAVITY = 0.5
SWING_SPEED = 0.1
MAX_ITERATIONS = 500

# Building anchors (x, y) - top of buildings
ANCHORS = [(100, 100), (250, 80), (400, 120), (550, 90), (700, 110)]

class SwingAgent:
    def __init__(self):
        self.x = 50.0
        self.y = 300.0
        self.vx = 5.0
        self.vy = 0.0
        self.anchor = None
        self.angle = 0.0
        self.angular_velocity = 0.0
        self.iterations = 0

    def observe(self, state):
        self.x = state['x']
        self.y = state['y']
        self.vx = state['vx']
        self.vy = state['vy']

    def decide(self):
        if self.anchor is None:
            # Find nearest anchor ahead within range
            best = None
            best_dist = float('inf')
            for ax, ay in ANCHORS:
                if ax > self.x and ax - self.x < 300:
                    dist = math.hypot(ax - self.x, ay - self.y)
                    if dist < best_dist:
                        best_dist = dist
                        best = (ax, ay)
            if best:
                self.anchor = best
                # Initialize angle from vertical
                dx = self.x - best[0]
                dy = self.y - best[1]
                self.angle = math.atan2(dx, dy)
                self.angular_velocity = 0.0
                return {'action': 'attach', 'anchor': best}
        else:
            # Pendulum physics
            length = math.hypot(self.x - self.anchor[0], self.y - self.anchor[1])
            if length < 1:
                length = 1
            alpha = -GRAVITY / length * math.sin(self.angle)
            self.angular_velocity += alpha * 0.1
            self.angle += self.angular_velocity * 0.1
            # Check release condition: at bottom and moving forward
            if abs(self.angle) < 0.1 and self.angular_velocity > 0:
                # Release
                self.anchor = None
                return {'action': 'release', 'vx': self.vx, 'vy': self.vy}
            else:
                # Update position from pendulum
                new_x = self.anchor[0] + length * math.sin(self.angle)
                new_y = self.anchor[1] + length * math.cos(self.angle)
                self.vx = (new_x - self.x) / 0.1
                self.vy = (new_y - self.y) / 0.1
                self.x = new_x
                self.y = new_y
                return {'action': 'swing', 'x': self.x, 'y': self.y}
        return {'action': 'none'}

    def should_terminate(self):
        return self.x > SCREEN_WIDTH or self.iterations > MAX_ITERATIONS

async def handler(websocket):
    agent = SwingAgent()
    async for message in websocket:
        state = json.loads(message)
        agent.observe(state)
        if agent.should_terminate():
            await websocket.send(json.dumps({'action': 'stop'}))
            break
        command = agent.decide()
        agent.iterations += 1
        await websocket.send(json.dumps(command))

async def main():
    async with websockets.serve(handler, "localhost", 8765):
        await asyncio.Future()  # run forever

if __name__ == "__main__":
    asyncio.run(main())

Browser Client (index.html)

The client renders a stick figure and sends state updates. It uses the Canvas API.

<!DOCTYPE html>
<html>
<head><title>Spider-Man Swing</title></head>
<body>
<canvas id="canvas" width="800" height="600"></canvas>
<script>
const canvas = document.getElementById('canvas');
const ctx = canvas.getContext('2d');
const ws = new WebSocket('ws://localhost:8765');

let state = { x: 50, y: 300, vx: 5, vy: 0 };
let anchor = null;

ws.onmessage = (event) => {
    const cmd = JSON.parse(event.data);
    if (cmd.action === 'attach') {
        anchor = cmd.anchor;
    } else if (cmd.action === 'release') {
        anchor = null;
    } else if (cmd.action === 'swing') {
        state.x = cmd.x;
        state.y = cmd.y;
    } else if (cmd.action === 'stop') {
        ws.close();
    }
};

function draw() {
    ctx.clearRect(0, 0, 800, 600);
    // Draw buildings
    ctx.fillStyle = '#333';
    [[100, 100], [250, 80], [400, 120], [550, 90], [700, 110]].forEach(([x, y]) => {
        ctx.fillRect(x - 20, y, 40, 600 - y);
    });
    // Draw web
    if (anchor) {
        ctx.beginPath();
        ctx.moveTo(state.x, state.y);
        ctx.lineTo(anchor[0], anchor[1]);
        ctx.strokeStyle = 'white';
        ctx.stroke();
    }
    // Draw Spider-Man (stick figure)
    ctx.beginPath();
    ctx.arc(state.x, state.y, 10, 0, Math.PI * 2);
    ctx.fillStyle = 'red';
    ctx.fill();
    ctx.beginPath();
    ctx.moveTo(state.x, state.y + 10);
    ctx.lineTo(state.x, state.y + 30);
    ctx.strokeStyle = 'red';
    ctx.stroke();
    // Send state
    ws.send(JSON.stringify(state));
    requestAnimationFrame(draw);
}

draw();
</script>
</body>
</html>

Running It

Save server.py and run python server.py.
Open index.html in a browser.
Watch Spider-Man swing. The agent loop terminates when he reaches the right edge or after 500 iterations.

Security warning: This code uses eval? No. It doesn’t. But if you extend it to execute arbitrary code from the browser, never use eval or exec on untrusted input. Always validate and sanitize. The WebSocket server here only accepts JSON with numeric fields; it does not execute code.

Key Takeaways

Agent loops can drive animation: You don’t need an LLM to have an observe-decide-act loop. A simple rule-based controller can produce emergent motion.
Explicit termination is critical: Without a stop condition, the agent would run forever. I used x > SCREEN_WIDTH and a max iteration cap.
Physics can be simple: A pendulum model with release timing is enough for a convincing swing.
Real-time constraints: The loop runs at 30 Hz; the browser interpolates. This keeps CPU usage low.
No frames were animated: The motion is generated procedurally from parameters, not from pre-rendered images.

This approach is extensible: you could swap the rule-based agent for a reinforcement learning policy, or add more complex environments. But even this minimal version shows how an agent-style loop can create dynamic, interactive experiences without traditional animation.

Your API's Newest Users Are Agents: Designing for Non-Human Clients

heavy shield — Wed, 07 Oct 2026 02:41:25 +0000

Introduction

Your API's newest users are agents. They don't read docs, they don't browse dashboards, and they don't file support tickets. They parse OpenAPI specs, call endpoints in loops, and expect deterministic, machine-readable responses. If your API was designed for humans clicking buttons, it's already failing this new class of client.

In this post, I'll walk through a concrete example: building a small "tool API" that an LLM-based agent can call. We'll cover the problem, a solution, and a runnable Python implementation. No hand-waving about "agentic workflows" — just code you can run and a loop with explicit termination conditions.

Problem

Human-facing APIs optimize for discoverability and forgiveness. Agents optimize for determinism and low token cost. Three specific failure modes show up when agents hit a human-designed API:

Ambiguous errors. A 400 with {"error": "bad request"} forces the agent to guess. The agent will retry, hallucinate a fix, or give up.
Unbounded responses. Returning 50KB of nested JSON burns context window and increases the chance the agent truncates the useful part.
No idempotency. Agents retry on timeouts. Without idempotency keys, you get duplicate side effects.

Here's a minimal example of the kind of handler that causes these problems:

# bad_handler.py
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/create_ticket", methods=["POST"])
def create_ticket_bad():
    data = request.get_json(silent=True) or {}
    if "title" not in data:
        return jsonify({"error": "bad request"}), 400
    # ... create ticket, no idempotency, returns everything
    return jsonify({"ticket": {"id": 1, "title": data["title"], "history": [...]}})

An agent calling this has no way to distinguish "missing field" from "malformed JSON" from "server error," and no safe retry path.

Solution

Design the API for a client that is literal, stateless between calls, and token-budgeted. Concretely:

Structured error codes. Every error response includes a stable code string the agent can branch on.
Idempotency keys. Accept an Idempotency-Key header; store the result keyed by it.
Bounded payloads. Return only fields the agent needs; paginate lists with explicit cursors.
A tool schema. Publish a JSON Schema your agent can load to construct calls. This is OpenAI-function-calling-style, not tied to any specific vendor.

Implementation

We'll build a small in-memory ticket API with the properties above, then write an agent loop that uses it. The agent loop is deliberately simple: it's a plan-act-observe loop with a hard step limit and a success predicate. No framework required.

The API

# agent_api.py
import json
import uuid
from dataclasses import dataclass, field, asdict
from typing import Optional

from flask import Flask, request, jsonify

app = Flask(__name__)

@dataclass
class Ticket:
    id: str
    title: str
    status: str = "open"

tickets: dict[str, Ticket] = {}
idempotency_store: dict[str, dict] = {}

TOOL_SCHEMA = {
    "name": "create_ticket",
    "description": "Create a support ticket. Idempotent on Idempotency-Key header.",
    "parameters": {
        "type": "object",
        "properties": {
            "title": {"type": "string", "minLength": 1, "maxLength": 200},
        },
        "required": ["title"],
        "additionalProperties": False,
    },
}

@app.route("/tools", methods=["GET"])
def list_tools():
    return jsonify({"tools": [TOOL_SCHEMA]})

@app.route("/create_ticket", methods=["POST"])
def create_ticket():
    key = request.headers.get("Idempotency-Key")
    if not key:
        return jsonify({"code": "missing_idempotency_key",
                        "message": "Provide Idempotency-Key header."}), 400

    if key in idempotency_store:
        return jsonify(idempotency_store[key]), 200

    data = request.get_json(silent=True)
    if not isinstance(data, dict):
        return jsonify({"code": "invalid_json",
                        "message": "Body must be a JSON object."}), 400

    title = data.get("title")
    if not isinstance(title, str) or not title.strip():
        return jsonify({"code": "invalid_title",
                        "message": "'title' must be a non-empty string."}), 422

    ticket = Ticket(id=str(uuid.uuid4()), title=title.strip())
    tickets[ticket.id] = ticket
    payload = {"ticket": asdict(ticket)}
    idempotency_store[key] = payload
    return jsonify(payload), 201

@app.route("/tickets/<ticket_id>", methods=["GET"])
def get_ticket(ticket_id: str):
    ticket = tickets.get(ticket_id)
    if ticket is None:
        return jsonify({"code": "not_found",
                        "message": f"No ticket {ticket_id}."}), 404
    return jsonify({"ticket": asdict(ticket)})

Key details: every error has a stable code; the response body is small; idempotency is enforced via header, not body, so retries are safe.

The Agent Loop

The agent loop below is deliberately framework-free. It defines explicit termination conditions: it stops when the goal is met, when the model returns no tool call, or when max_steps is reached. I'm using an OpenAI-chat-completions-style interface here as a stand-in; swap in whichever client you use.

# agent_loop.py
import json
import uuid
from typing import Any

import requests

API = "http://localhost:5000"
MAX_STEPS = 6


def call_model(messages: list[dict]) -> dict:
    """Return a message dict. Replace with your model client."""
    # Placeholder: in production, call your LLM here.
    raise NotImplementedError("Wire up your model client.")


def call_tool(name: str, args: dict) -> dict:
    if name == "create_ticket":
        r = requests.post(
            f"{API}/create_ticket",
            json=args,
            headers={"Idempotency-Key": str(uuid.uuid4())},
            timeout=10,
        )
        return {"status": r.status_code, "body": r.json()}
    return {"status": 400, "body": {"code": "unknown_tool", "message": name}}


def run_agent(goal: str) -> dict:
    tools = requests.get(f"{API}/tools", timeout=10).json()["tools"]
    messages = [
        {"role": "system", "content": "You are an agent. Use tools to satisfy the goal."},
        {"role": "user", "content": goal},
    ]

    for step in range(MAX_STEPS):
        msg = call_model(messages)
        messages.append(msg)

        # Termination condition 1: model produced a final answer.
        if not msg.get("tool_calls"):
            return {"status": "done", "steps": step, "answer": msg.get("content")}

        for call in msg["tool_calls"]:
            name = call["function"]["name"]
            args = json.loads(call["function"]["arguments"])
            result = call_tool(name, args)
            messages.append({
                "role": "tool",
                "tool_call_id": call["id"],
                "content": json.dumps(result),
            })

    # Termination condition 2: step budget exhausted.
    return {"status": "max_steps_reached", "steps": MAX_STEPS}

The loop has exactly two exit paths, both explicit. There's no "keep trying until it works" branch, which is where most agent bugs live.

Security Notes

If you extend this to execute code or shell commands, treat the tool boundary as a trust boundary. Never pass model output to eval, exec, or subprocess with shell=True. If you must run generated code, isolate it in a sandbox (a container with no network, a read-only filesystem, and a hard timeout) and validate arguments against the JSON Schema before execution. The example above avoids this entirely by only allowing a single typed tool with a bounded string parameter.

Key Takeaways

Agents are literal clients. Stable error code fields beat prose messages.
Idempotency keys belong in headers and should be required for any write endpoint an agent can retry.
Publish a JSON Schema for your tools so the agent can construct calls without guessing.
Define agent loops with explicit termination conditions — a step limit and a success predicate. "Loop until done" is not a termination condition.
Keep payloads small. Every byte you return is a token the agent pays for.
If you add code-execution tools, sandbox them. eval on model output is a remote code execution vulnerability with extra steps.

The code above is intentionally minimal so you can fork it and add your own tools. Start by auditing your existing API for the three failure modes in the Problem section — that's where the real work is.

What If Your AI Agent Never Had to Leave the Browser?

heavy shield — Tue, 29 Sep 2026 05:30:38 +0000

Introduction

Most AI agents today run in a Python process on a server or your laptop. They call APIs, maybe execute shell commands, and return text. But what if the agent's entire runtime lived inside a browser tab? No backend, no container, no SSH. Just JavaScript and WebAssembly, with a Python kernel compiled to WASM.

This post walks through a minimal agent loop that runs entirely client-side using Pyodide — Python in the browser via WebAssembly. We'll build a tool-using agent that can do arithmetic, read from a virtual filesystem, and stop under explicit conditions. All code is runnable in a modern browser.

Problem

Server-side agents have friction:

Deployment: You need a host, secrets management, and network egress.
Latency: Every tool call is a round trip.
Security surface: Arbitrary code execution on a server is a serious risk.
Demo friction: Sharing a working agent means sharing infrastructure.

A browser-native agent flips this. The sandbox is the browser tab. The runtime is WebAssembly. The only network call is loading the Python runtime itself.

Solution

We'll use Pyodide to run CPython in the browser. The agent loop is a simple ReAct-style loop: the model proposes a tool call, we execute it in Python, append the result, and repeat until a termination condition is met.

We won't call a real LLM here — instead we use a deterministic policy function so the demo is reproducible and offline. Swap the policy for a fetch to an LLM API and the loop is unchanged. We'll call this a "ReAct-style" loop because it follows the observe-think-act pattern, not because it implements the exact paper.

Termination conditions (explicit):

The policy returns a final action.
The step count exceeds MAX_STEPS (default 8).
A tool raises an exception that the policy cannot recover from.

Implementation

1. HTML shell

<!doctype html>
<html>
<head><meta charset="utf-8"><title>Browser Agent</title></head>
<body>
  <pre id="log"></pre>
  <script src="https://cdn.jsdelivr.net/pyodide/v0.26.2/full/pyodide.js"></script>
  <script type="module">
    const log = (m) => document.getElementById('log').textContent += m + '\n';
    const pyodide = await loadPyodide();
    await pyodide.runPythonAsync(await (await fetch('agent.py')).text());
    const result = await pyodide.runPythonAsync('run_agent("What is 21 * 2 plus 8?")');
    log(result);
  </script>
</body>
</html>

2. The agent module (`agent.py`)

import json
import re

MAX_STEPS = 8

# --- Tools -------------------------------------------------------------

def tool_calc(expr: str) -> str:
    """Evaluate a pure arithmetic expression.

    SECURITY WARNING: eval() executes arbitrary Python. This implementation
    restricts input to digits and operators via a regex whitelist. Do not
    remove the whitelist. Do not pass user-controlled strings from an
    untrusted source. For production, use ast.literal_eval or a parser.
    """
    if not re.fullmatch(r"[0-9+\-*/(). ]+", expr):
        raise ValueError(f"unsafe expression: {expr!r}")
    return str(eval(expr, {"__builtins__": {}}, {}))


_FS = {"notes.txt": "remember: 42"}

def tool_read_file(path: str) -> str:
    if path not in _FS:
        raise FileNotFoundError(path)
    return _FS[path]


TOOLS = {
    "calc": tool_calc,
    "read_file": tool_read_file,
}

# --- Policy (replace with an LLM call) ---------------------------------

def policy(question: str, history: list) -> dict:
    """Deterministic stand-in for an LLM. Returns a dict action.

    Swap this for a fetch() to your LLM of choice. The loop below does not
    care where the action came from.
    """
    if not history:
        # First turn: extract a math expression from the question.
        m = re.search(r"([0-9+\-*/(). ]+)", question)
        if m:
            return {"type": "tool", "name": "calc", "args": {"expr": m.group(1).strip()}}
        return {"type": "final", "content": "no expression found"}

    last = history[-1]
    if last["role"] == "tool" and last["name"] == "calc":
        # Second turn: add 8 as required by the question.
        return {"type": "tool", "name": "calc", "args": {"expr": f"{last['result']} + 8"}}
    if last["role"] == "tool" and last["name"] == "calc":
        return {"type": "final", "content": last["result"]}
    return {"type": "final", "content": "done"}

# --- Agent loop --------------------------------------------------------

def run_agent(question: str) -> str:
    history = []
    for step in range(MAX_STEPS):
        action = policy(question, history)
        history.append({"role": "assistant", "action": action})

        if action["type"] == "final":
            return f"[step {step}] {action['content']}"

        if action["type"] == "tool":
            fn = TOOLS.get(action["name"])
            if fn is None:
                history.append({"role": "tool", "name": action["name"],
                                "error": "unknown tool"})
                continue
            try:
                result = fn(**action["args"])
                history.append({"role": "tool", "name": action["name"],
                                "result": result})
            except Exception as e:
                history.append({"role": "tool", "name": action["name"],
                                "error": str(e)})

    return f"[halted: exceeded MAX_STEPS={MAX_STEPS}]"

3. What actually happens

The browser loads Pyodide (about 6 MB gzipped).
agent.py is fetched and executed in the WASM interpreter.
run_agent runs the loop. Step 0 calls calc("21 * 2") → 42. Step 1 calls calc("42 + 8") → 50. Step 2 returns final.
The result is printed to the page.

No server. No API key. The entire agent state lives in the tab and is discarded when you close it.

4. Swapping in a real LLM

Replace policy with something like:

import json
from js import fetch  # Pyodide exposes the browser fetch API

async def policy_llm(question, history):
    resp = await fetch(
        "https://api.example.com/v1/chat",
        {"method": "POST",
         "headers": {"Content-Type": "application/json"},
         "body": json.dumps({"q": question, "history": history})}
    )
    data = await resp.json()
    return json.loads(data.action)

You'll need to make run_agent async and await the policy. The rest of the loop is identical. Note that calling an LLM from the browser exposes your API key to the user — use a short-lived token or a proxy you control.

Key Takeaways

Browser-native agents are real. Pyodide gives you CPython in a tab. That's enough to run a tool-using loop, a virtual filesystem, and a policy function.
Explicit termination matters. Define MAX_STEPS, a final action, and error handling. Without them, loops run forever and burn tokens.
The sandbox is the feature. No server means no remote code execution surface, but it also means no secrets. Design accordingly.
Never eval untrusted input. The tool_calc example uses a regex whitelist and an empty __builtins__ map. That is a mitigation, not a guarantee. For anything real, use ast.literal_eval or a dedicated parser.
The policy is swappable. The loop is agnostic to whether the action came from a regex, a local model, or a remote LLM. That separation is what makes the pattern portable.

Try it: drop the two files in a folder, serve with python -m http.server, and open the page. You'll have an agent that never leaves the browser.

Running an AI Agent Inside the Browser with Pyodide and Ollama

heavy shield — Mon, 28 Sep 2026 00:10:35 +0000

Introduction

Browser automation is the default answer when an AI agent needs to use a web app: launch headless Chrome, drive it with Playwright, ferry state back and forth. It works, but the agent ends up living outside the browser, re-implementing sessions and rendering in parallel.

There's a less obvious option: run the agent inside the browser tab. Pyodide gives us CPython in WebAssembly, the page's DOM and network are directly reachable, and a local model such as Ollama can act as the reasoning backend. This post walks through a small working experiment along those lines, inspired by the tool-call pattern of MCP (Model Context Protocol).

One clarification up front, because it matters: what we build here is an MCP-style tool layer, not a full MCP protocol implementation. The official MCP SDK defines tools, resources, and prompts over standard transports (stdio, Streamable HTTP, SSE). We borrow the "model calls named tools with structured arguments" idea and implement it with a simple JSON message shape. If you want wire-level MCP in the browser, that's a follow-up project, not this one.

The Problem

Driving a browser from Python looks like this:

from playwright.sync_api import sync_playwright

def get_page_text(url):
    with sync_playwright() as p:
        browser = p.chromium.launch()
        page = browser.new_page()
        page.goto(url)
        text = page.inner_text("body")
        browser.close()
        return text

Every interaction is a round-trip between two processes. Session state (logins, cookies, rendered state) lives in the browser your user is actually looking at, while the agent operates on a separate, synthetic copy. Keeping those two worlds in sync is where the bugs and the latency come from.

The Solution

Flip the arrangement: the agent loop runs in the tab, and the page's own APIs become its tools. The architecture:

Browser tab runs a Python agent loop on Pyodide (CPython compiled to WebAssembly).
A small tool registry in the same tab exposes DOM operations (read_dom, click_element) as named tools the model can call with JSON arguments — the same shape MCP popularized.
A local model (Ollama) receives each tool call and returns the next action.
Results are executed directly against the live page.

Model inference does not happen in the browser — the agent loop does. The division of labor is: reasoning in Ollama, perception and action in the tab.

Implementation

Step 1: A tool registry in the browser

Create tools.py:

import json
import js

class ToolRegistry:
    """MCP-style named tools, callable via a simple JSON protocol."""

    def __init__(self):
        self.tools = {
            "read_dom": self.read_dom,
            "click_element": self.click_element,
            "evaluate_js": self.evaluate_js,
            "done": self.done,
        }

    def read_dom(self, selector="body"):
        element = js.document.querySelector(selector)
        if not element:
            return f"No element found for selector: {selector}"
        return element.innerText

    def click_element(self, selector):
        element = js.document.querySelector(selector)
        if not element:
            return f"No element found for selector: {selector}"
        element.click()
        return f"Clicked {selector}"

    def done(self, message=""):
        """Signal that the task is complete."""
        return f"TASK_DONE: {message}"

    def evaluate_js(self, code):
        try:
            return str(js.eval(code))
        except Exception as e:
            return f"JS error: {e}"

    def handle_request(self, request_json):
        request = json.loads(request_json)
        name, args = request.get("tool"), request.get("args", {})
        if name not in self.tools:
            return json.dumps({"error": f"Unknown tool: {name}"})
        try:
            return json.dumps({"result": self.tools[name](**args)})
        except Exception as e:
            return json.dumps({"error": str(e)})

registry = ToolRegistry()
js.window.toolRegistry = registry

Security note: evaluate_js deliberately exposes arbitrary JavaScript execution, including access to the page's session and credentials context. That's acceptable for a local experiment on pages you own. A real deployment should restrict the operation set (whitelist specific actions) and never expose unrestricted eval — the same way a production MCP server validates and sanitizes every tool input.

Step 2: Agent loop in Python

Create agent.py:

import json
import js
from pyodide.http import pyfetch

async def call_ollama(prompt):
    response = await pyfetch(
        "http://localhost:11434/api/generate",
        method="POST",
        headers={"Content-Type": "application/json"},
        body=json.dumps({
            "model": "llama3.2",
            "prompt": prompt,
            "stream": False,
            "format": "json",
        }),
    )
    data = await response.json()
    return data["response"]

async def agent_step(task):
    dom_text = js.window.toolRegistry.read_dom("body")
    prompt = f"""You are a browser agent. Task: {task}
Current page content:
{dom_text[:2000]}

Respond with JSON: {{"tool": "read_dom"|"click_element"|"evaluate_js"|"done", "args": {{...}}}}
Use "done" when the task is complete.
"""
    action = json.loads(await call_ollama(prompt))
    return js.window.toolRegistry.handle_request(json.dumps(action))

async def run_agent(task, max_steps=5):
    for _ in range(max_steps):
        result = await agent_step(task)
        print("Result:", result)
        if "TASK_DONE" in result:
            break

Two things have to be true for this to work:

Ollama must accept cross-origin requests. Browser code calling http://localhost:11434 is a cross-origin request, and Ollama rejects those by default. Start it with an allowlist: OLLAMA_ORIGINS="http://localhost:8000" ollama serve (use your page's actual origin; * works for local experiments only).
The page must reach Ollama over HTTP, not mixed content. Serve your page from http://localhost rather than opening it as file://, otherwise some browsers block the call.

Step 3: Load Pyodide and run

In your HTML page:

<script src="https://cdn.jsdelivr.net/pyodide/v0.26.2/full/pyodide.js"></script>
<script>
async function main() {
  const pyodide = await loadPyodide();
  await pyodide.runPythonAsync(await (await fetch('tools.py')).text());
  await pyodide.runPythonAsync(await (await fetch('agent.py')).text());
  await pyodide.runPythonAsync("import asyncio; asyncio.ensure_future(run_agent('find the pricing link'))");
}
main();
</script>

Run any static file server (python -m http.server 8000), start Ollama with the origins allowlist, open the page, and watch the tab operate on itself.

Key Takeaways

Running the agent loop in the tab removes the browser-automation glue layer: no subprocess management, no session replication.
The tool-call pattern (model emits structured tool invocations, host executes them) travels well beyond MCP's wire protocol — a plain JSON message shape is enough for experiments.
The division of labor matters: agent loop in WebAssembly, inference on a local Ollama. Claiming "everything runs in the browser" would ignore the CORS and mixed-content constraints above.
Unrestricted eval as a model-callable tool is a local-experiment-only feature. Production versions whitelist operations and validate arguments.