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:
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:
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 and our developer community @ossairu. If you are building autonomous agent infrastructure, feel free to join.