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

> Source: <https://dev.to/m1kulya/stateful-vs-stateless-mcp-why-your-ai-agents-crash-and-how-to-build-a-resilient-harness-for-zsh-3d5k>
> Published: 2026-09-22 10:57:40+00:00

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:

`sqlite3.OperationalError: database is locked`.` 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`:

``` python
# 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**:

``` python
# 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()
```

`--help` on demand)`127.0.0.1`.` systemd` to maintain 100% uptime.
*I regularly share architecture blueprints, production LaunchAgent manifests, and self-hosted AI alternatives in **[OpenSource AI Radar](https://t.me/ossairadar)** and our developer community **[@ossairu](https://t.me/ossairu)**. If you are building autonomous agent infrastructure, feel free to join.*