Stateful vs Stateless MCP: Why Your AI Agents Crash and How to Build a Resilient Harness for zsh

When you connect the first few tools to an AI agent (Cursor, Claude Code, Antigravity, or a custom script) using the Model Context Protocol (MCP), it feels like magic. You define stdio servers in your config, and the agent reads files, queries databases, and runs shell commands seamlessly. However, once your setup scales past 10+ tools and includes stateful services (Telegram MTProto sessions, authenticated Chrome instances, persistent PostgreSQL connection pools, or background workers), things quickly fall apart: SessionLock and Zombie Processes: Every time your IDE or subagent restarts, it spawns a duplicate child process that attempts to acquire an exclusive lock on the local SQLite session file, crashing with sqlite3.OperationalError: database is locked. Context Window Explosion (Token Bleed): Registering 30+ tools directly via MCP JSON Schema eats 8,000 to 15,000 tokens on every single prompt before you even type your request. Broken Pipe Failures: Long-running background jobs trigger agent timeouts, causing the agent runtime to abruptly sever stdin/stdout pipes. In this guide, we break down how to design a production-grade AI agent harness: separating MCP into Stateless and Stateful layers, deploying local Streamable HTTP daemons under a system supervisor (launchd on macOS or systemd on Linux), and slashing token overhead by 80% using compact CLI wrappers. stdio Destroys Stateful Services By default, MCP promotes stdio process spawning: { "mcpServers": { "telegram": { "command": "python3", "args": ["/path/to/telegram_server.py"] } } } For purely stateless utilities (like a currency converter, read-only file viewer, or git diff), this is completely fine. The process starts, handles the JSON-RPC request over standard I/O, outputs the response, and exits. Now, consider what happens with a stateful client like Telegram (via Telethon or Pyrogram): Telethon stores cryptographic session keys inside a local SQLite database (anon.session). When your agent runtime restarts or launches concurrent subagents, a second Python process starts up and tries to access the exact same database. It immediately hits a lock contention, crashes, and leaves a dangling orphan process in your operating system. To achieve rock-solid reliability, we split our tool stack into two distinct tiers: Metric Stateless (stdio) Stateful (Streamable HTTP Daemon) Startup Overhead Slow (cold-starts Python interpreter every invocation) Instant (daemon is pre-warmed in memory) Socket & Network State Dropped on session exit Maintained 24/7 in background Concurrent Access Dangerous (risk of file corruptions & locks) Safe (synchronized via asyncio event loop) Resource Footprint N processes per N subagents 1 single lightweight daemon for the entire machine Let's convert our stateful service into a long-running daemon supporting Server-Sent Events (SSE) and Streamable HTTP using Python's FastMCP: # tg_daemon.py import asyncio import os from contextlib import asynccontextmanager from mcp.server.fastmcp import FastMCP from telethon import TelegramClient API_ID = int(os.environ["TG_API_ID"]) API_HASH = os.environ["TG_API_HASH"] SESSION_PATH = os.path.expanduser("~/.local/share/tg_session/client.session") client = TelegramClient(SESSION_PATH, API_ID, API_HASH) @asynccontextmanager async def lifespan(app): await client.connect() if not await client.is_user_authorized(): raise RuntimeError("Session not authorized. Run auth helper first.") yield await client.disconnect() mcp = FastMCP("TelegramStatefulService", lifespan=lifespan) @mcp.tool() async def send_broadcast(chat_id: str, message: str) -> str: """Send message to target chat without reconnecting""" entity = await client.get_entity(chat_id) sent = await client.send_message(entity, message) return f"Message delivered successfully. ID: {sent.id}" @mcp.tool() async def fetch_unread_summary(limit: int = 10) -> str: """Read unread counters from warm connection""" dialogs = await client.get_dialogs(limit=limit) unreads = [f"{d.name}: {d.unread_count} unread" for d in dialogs if d.unread_count > 0] return "\n".join(unreads) if unreads else "No unread messages" if __name__ == "__main__": mcp.run(transport="sse", host="127.0.0.1", port=8765) Now the daemon holds a single, warm connection to Telegram servers. It never drops socket connections and is 100% immune to SQLite locking bugs. In your agent's MCP configuration, point to the local HTTP endpoint: { "mcpServers": { "telegram": { "url": "http://127.0.0.1:8765/sse" } } } launchd To ensure the daemon starts on boot and restarts automatically upon any system crash, register it as a user-level daemon in ~/Library/LaunchAgents/com.mika.tg-mcp.plist: <?xml version="1.0" encoding="UTF-8"?> <!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd"> <plist version="1.0"> <dict> <key>Label</key> <string>com.mika.tg-mcp</string> <key>ProgramArguments</key> <array> <string>/Users/mika/.venv/bin/python3</string> <string>/Users/mika/Project/HarnessSetup/bin/tg_daemon.py</string> </array> <key>RunAtLoad</key> <true/> <key>KeepAlive</key> <dict> <key>SuccessfulExit</key> <false/> <key>NetworkState</key> <true/> </dict> <key>StandardOutPath</key> <string>/Users/mika/.local/share/logs/tg_mcp.log</string> <key>StandardErrorPath</key> <string>/Users/mika/.local/share/logs/tg_mcp_err.log</string> </dict> </plist> Activate the service: launchctl load ~/Library/LaunchAgents/com.mika.tg-mcp.plist It consumes ~25 MB of RAM, stays active 24/7, and responds instantly. Exposing 35 separate tools with verbose JSON Schema signatures consumes 10,000+ tokens on every single agent iteration. Instead of registering dozens of raw functions, we provide the agent with a single unified CLI wrapper: # tg_cli.py import argparse import httpx DAEMON_URL = "http://127.0.0.1:8765" def main(): parser = argparse.ArgumentParser(description="Compact Telegram CLI for AI Agents") subparsers = parser.add_subparsers(dest="command") send_p = subparsers.add_parser("send", help="Send message") send_p.add_argument("--to", required=True) send_p.add_argument("--text", required=True) subparsers.add_parser("unreads", help="Fetch unreads") args = parser.parse_args() if args.command == "send": r = httpx.post(f"{DAEMON_URL}/tools/send_broadcast", json={"chat_id": args.to, "message": args.text}) print(r.text) elif args.command == "unreads": r = httpx.post(f"{DAEMON_URL}/tools/fetch_unread_summary", json={}) print(r.text) else: parser.print_help() if __name__ == "__main__": main() Direct MCP (30 tools): ~12,500 tokens per prompt HTTP Daemon + CLI wrapper: ~150 tokens per prompt (reads --help on demand) Cost reduction: Over 80% savings on API billing across long reasoning sessions Never use stdio for stateful services. If a tool holds active database connections, sockets, browser profiles, or auth tokens, run it as an isolated HTTP daemon on 127.0.0.1. Let the OS supervise processes. Use launchd or systemd to maintain 100% uptime. Optimize your token budget. Reserve direct MCP declarations for 3–5 core tools and wrap large toolsets in CLI interfaces. I regularly share architecture blueprints, production LaunchAgent manifests, and self-hosted AI alternatives in *OpenSource AI Radar** and our developer community @ossairu. If you are building autonomous agent infrastructure, feel free to join.*