Replace JSON timeline/A2A stores with shared SQL persistence (SQLite/Postgres)

.env.example

@@ -11,6 +11,13 @@ X_USERNAME=your_twitter_username
 # Grok/xAI Settings
 XAI_API_KEY=your_xai_api_key_here
 
+# Timeline/A2A persistence (Python services)
+# Local SQLite example:
+# DATABASE_URL=sqlite:///~/.xmcp/xmcp.db
+# Railway Postgres example (also accepts postgres:// and auto-normalizes):
+# DATABASE_URL=******host:5432/dbname
+DATABASE_URL=
+
 # Agent Settings
 POLLING_INTERVAL_MS=30000
 MAX_RETRIES=3

README.md

@@ -166,13 +166,24 @@ Wire the cross-service URLs after deployment:
 ```
 MCP_SERVER_URL=https://<mcp-server>.up.railway.app/mcp
 TIMELINE_API_URL=https://<timeline-server>.up.railway.app
+# All four services must share one database:
+DATABASE_URL=postgres://<user>:<pass>@<host>:<port>/<db>
 ```
 
 See `docs/DEPLOYMENT.md` for full Railway setup details.
 
 ## Data Storage
 
-Timeline cards and A2A messages are stored in JSON files at `~/.xmcp/` by default. This is intentional for lightweight local use. For production, override `TIMELINE_STORE_PATH` and `A2A_STORE_PATH` to point to a persistent volume.
+Timeline cards and A2A messages are stored in SQL and selected by `DATABASE_URL`:
+
+- `DATABASE_URL` unset (or `sqlite://...`) → SQLite (`~/.xmcp/xmcp.db` by default)
+- `postgres://...` or `postgresql://...` → Postgres (Railway-ready; `postgres://` is normalized automatically)
+
+Tables are created automatically on startup. For local concurrency safety, SQLite enables WAL mode and a busy timeout. To migrate legacy JSON stores (`TIMELINE_STORE_PATH`, `A2A_STORE_PATH`), run:
+
+```bash
+python scripts/migrate_json_to_sql.py
+```
 
 ## Project Structure
 
@@ -181,8 +192,10 @@ Timeline cards and A2A messages are stored in JSON files at `~/.xmcp/` by defaul
 ├── timeline_server.py     # Timeline + A2A FastAPI server
 ├── listener.py            # X mention poller + Grok responder
 ├── mcp_dispatcher.py      # Timeline action executor
-├── timeline_store.py      # JSON-file timeline persistence
-├── a2a_store.py           # JSON-file A2A persistence
+├── timeline_store.py      # SQL-backed timeline persistence
+├── a2a_store.py           # SQL-backed A2A persistence
+├── storage_db.py          # Shared SQLAlchemy engine/schema
+├── scripts/migrate_json_to_sql.py
 ├── openapi.json           # X API OpenAPI spec (used by MCP server)
 ├── src/                   # TypeScript standalone agent (alternative)
 ├── docs/                  # Architecture, deployment, usage guides

a2a_store.py

@@ -1,13 +1,17 @@
-import json
-import os
-import threading
 import uuid
-from datetime import datetime, timezone
-from pathlib import Path
 from typing import Any, Dict, List, Optional
 
-A2A_STORE_PATH = Path(os.getenv("A2A_STORE_PATH", "~/.xmcp/a2a_store.json")).expanduser()
-A2A_STORE_LOCK = threading.Lock()
+from sqlalchemy import insert, select, update
+
+from storage_db import (
+    a2a_agents,
+    a2a_messages,
+    row_to_dict,
+    serialize_record,
+    utc_now,
+    write_connection,
+    read_connection,
+)
 
 DEFAULT_AGENTS = [
     {
@@ -16,6 +20,7 @@
         "description": "Dispatches timeline actions to MCP-enabled tools.",
         "status": "online",
         "endpoint": "local",
+        "kind": "agent",
         "tags": ["mcp", "orchestrator"],
     },
     {
@@ -24,6 +29,7 @@
         "description": "Handles @mentions and X actions.",
         "status": "online",
         "endpoint": "x",
+        "kind": "agent",
         "tags": ["x", "social"],
     },
     {
@@ -32,53 +38,63 @@
         "description": "Flokk timeline surface.",
         "status": "online",
         "endpoint": "flokk",
+        "kind": "agent",
         "tags": ["ui", "timeline"],
     },
 ]
 
 
-def _utc_now() -> str:
-    return datetime.now(timezone.utc).isoformat()
+def _normalize_kind(value: Any) -> str:
+    return value if value in ("agent", "bot") else "agent"
+
 
+def _serialize_agent(record: Dict[str, Any]) -> Dict[str, Any]:
+    value = serialize_record(record)
+    value["kind"] = _normalize_kind(value.get("kind"))
+    return value
 
-def _ensure_store() -> None:
-    if A2A_STORE_PATH.exists():
-        return
-    A2A_STORE_PATH.parent.mkdir(parents=True, exist_ok=True)
-    data = {"agents": DEFAULT_AGENTS, "messages": []}
-    A2A_STORE_PATH.write_text(json.dumps(data, indent=2), encoding="utf-8")
 
+def _serialize_message(record: Dict[str, Any]) -> Dict[str, Any]:
+    value = serialize_record(record)
+    value["from"] = value.pop("from_agent")
+    value["to"] = value.pop("to_agent")
+    return value
 
-def _read_store() -> Dict[str, Any]:
-    _ensure_store()
-    raw = A2A_STORE_PATH.read_text(encoding="utf-8")
-    try:
-        data = json.loads(raw)
-    except json.JSONDecodeError:
-        data = {"agents": DEFAULT_AGENTS, "messages": []}
-    if "agents" not in data or not isinstance(data["agents"], list):
-        data["agents"] = DEFAULT_AGENTS
-    if "messages" not in data or not isinstance(data["messages"], list):
-        data["messages"] = []
-    return data
 
+def _ensure_default_agents() -> None:
+    with write_connection() as conn:
+        existing_ids = {
+            row[0]
+            for row in conn.execute(select(a2a_agents.c.id).where(a2a_agents.c.id.in_([a["id"] for a in DEFAULT_AGENTS])))
+        }
 
-def _write_store(data: Dict[str, Any]) -> None:
-    A2A_STORE_PATH.parent.mkdir(parents=True, exist_ok=True)
-    A2A_STORE_PATH.write_text(json.dumps(data, indent=2), encoding="utf-8")
+        for agent in DEFAULT_AGENTS:
+            if agent["id"] in existing_ids:
+                continue
+            conn.execute(
+                insert(a2a_agents).values(
+                    **agent,
+                    created_at=utc_now(),
+                )
+            )
 
 
 def list_agents() -> List[Dict[str, Any]]:
-    with A2A_STORE_LOCK:
-        return _read_store()["agents"]
+    _ensure_default_agents()
+    query = select(a2a_agents).order_by(a2a_agents.c.created_at.asc(), a2a_agents.c.id.asc())
+    with read_connection() as conn:
+        rows = conn.execute(query).fetchall()
+    return [_serialize_agent(row_to_dict(row)) for row in rows]
 
 
 def get_agent(agent_id: str) -> Optional[Dict[str, Any]]:
-    with A2A_STORE_LOCK:
-        for agent in _read_store()["agents"]:
-            if agent.get("id") == agent_id:
-                return agent
-    return None
+    _ensure_default_agents()
+    query = select(a2a_agents).where(a2a_agents.c.id == agent_id)
+    with read_connection() as conn:
+        row = conn.execute(query).fetchone()
+    if not row:
+        return None
+    return _serialize_agent(row_to_dict(row))
 
 
 def register_agent(payload: Dict[str, Any]) -> Dict[str, Any]:
@@ -88,46 +104,51 @@ def register_agent(payload: Dict[str, Any]) -> Dict[str, Any]:
         "description": payload.get("description", ""),
         "status": payload.get("status", "offline"),
         "endpoint": payload.get("endpoint", ""),
-        # Classification: "agent" (interactive, LLM-backed, autonomous) vs
-        # "bot" (deterministic function executor). Normalized so null or
-        # unknown values can never enter the registry.
-        "kind": payload.get("kind") if payload.get("kind") in ("agent", "bot") else "agent",
+        "kind": _normalize_kind(payload.get("kind")),
         "tags": payload.get("tags", []),
-        "created_at": _utc_now(),
     }
-    with A2A_STORE_LOCK:
-        data = _read_store()
-        for existing in data["agents"]:
-            if existing.get("id") == agent["id"]:
-                # Re-registration updates mutable fields so seeded records
-                # (e.g. x-agent without kind) don't stay stale forever.
-                for field in ("name", "description", "status", "endpoint", "kind", "tags"):
-                    existing[field] = agent[field]
-                _write_store(data)
-                return existing
-        data["agents"].append(agent)
-        _write_store(data)
-    return agent
+
+    with write_connection() as conn:
+        existing = conn.execute(
+            select(a2a_agents).where(a2a_agents.c.id == agent["id"])
+        ).fetchone()
+
+        if existing:
+            conn.execute(
+                update(a2a_agents)
+                .where(a2a_agents.c.id == agent["id"])
+                .values(**agent)
+            )
+            row = row_to_dict(existing)
+            row.update(agent)
+            return _serialize_agent(row)
+
+        created = {**agent, "created_at": utc_now()}
+        conn.execute(insert(a2a_agents).values(**created))
+        return _serialize_agent(created)
 
 
 def list_messages(agent_id: str) -> List[Dict[str, Any]]:
-    with A2A_STORE_LOCK:
-        data = _read_store()
-        return [msg for msg in data["messages"] if msg.get("to") == agent_id]
+    query = (
+        select(a2a_messages)
+        .where(a2a_messages.c.to_agent == agent_id)
+        .order_by(a2a_messages.c.created_at.desc())
+    )
+    with read_connection() as conn:
+        rows = conn.execute(query).fetchall()
+    return [_serialize_message(row_to_dict(row)) for row in rows]
 
 
 def add_message(payload: Dict[str, Any]) -> Dict[str, Any]:
     message = {
         "id": payload.get("id") or str(uuid.uuid4()),
-        "from": payload.get("from", "system"),
-        "to": payload.get("to", "timeline-ui"),
+        "from_agent": payload.get("from", "system"),
+        "to_agent": payload.get("to", "timeline-ui"),
         "type": payload.get("type", "info"),
         "content": payload.get("content", ""),
         "metadata": payload.get("metadata", {}),
-        "created_at": _utc_now(),
+        "created_at": utc_now(),
     }
-    with A2A_STORE_LOCK:
-        data = _read_store()
-        data["messages"].insert(0, message)
-        _write_store(data)
-    return message
+    with write_connection() as conn:
+        conn.execute(insert(a2a_messages).values(**message))
+    return _serialize_message(message)

docker-compose.yml

@@ -1,4 +1,19 @@
 services:
+  postgres:
+    image: postgres:16-alpine
+    environment:
+      - POSTGRES_DB=xmcp
+      - POSTGRES_USER=xmcp
+      - POSTGRES_HOST_AUTH_METHOD=trust
+    ports:
+      - "5432:5432"
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U xmcp -d xmcp"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+    restart: unless-stopped
+
   mcp-server:
     build: .
     command: python server.py
@@ -8,6 +23,10 @@ services:
     environment:
       - MCP_HOST=0.0.0.0
       - MCP_PORT=8000
+      - DATABASE_URL=postgresql://xmcp@postgres:5432/xmcp
+    depends_on:
+      postgres:
+        condition: service_healthy
     restart: unless-stopped
 
   timeline-server:
@@ -19,8 +38,12 @@ services:
     environment:
       - TIMELINE_HOST=0.0.0.0
       - TIMELINE_PORT=8080
+      - DATABASE_URL=postgresql://xmcp@postgres:5432/xmcp
     volumes:
       - xmcp-data:/root/.xmcp
+    depends_on:
+      postgres:
+        condition: service_healthy
     restart: unless-stopped
     healthcheck:
       test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
@@ -35,13 +58,16 @@ services:
     environment:
       - MCP_SERVER_URL=http://mcp-server:8000/mcp
       - TIMELINE_API_URL=http://timeline-server:8080
+      - DATABASE_URL=postgresql://xmcp@postgres:5432/xmcp
     volumes:
       - xmcp-data:/root/.xmcp
     depends_on:
       timeline-server:
         condition: service_healthy
       mcp-server:
         condition: service_started
+      postgres:
+        condition: service_healthy
     restart: unless-stopped
 
   mcp-dispatcher:
@@ -51,13 +77,16 @@ services:
     environment:
       - MCP_SERVER_URL=http://mcp-server:8000/mcp
       - TIMELINE_API_URL=http://timeline-server:8080
+      - DATABASE_URL=postgresql://xmcp@postgres:5432/xmcp
     volumes:
       - xmcp-data:/root/.xmcp
     depends_on:
       timeline-server:
         condition: service_healthy
       mcp-server:
         condition: service_started
+      postgres:
+        condition: service_healthy
     restart: unless-stopped
 
 volumes:

docs/DEPLOYMENT.md

@@ -2,6 +2,29 @@
 
 ## Deployment Options
 
+### Railway (Python service stack)
+
+Deploy **four services** from this same repository:
+
+| Service | Start command |
+|---|---|
+| `mcp-server` | `python server.py` |
+| `timeline-server` | `python timeline_server.py` |
+| `listener` | `python listener.py` |
+| `mcp-dispatcher` | `python mcp_dispatcher.py` |
+
+Add a Railway Postgres plugin, then set the **same** `DATABASE_URL` on all four services.
+
+Storage backend selection:
+- `DATABASE_URL` unset/`sqlite://...` → local SQLite (`~/.xmcp/xmcp.db`)
+- `postgres://...` or `postgresql://...` → Postgres (`postgres://` is normalized automatically)
+
+The timeline and A2A schema are auto-created at startup. If you are migrating old local JSON stores, run:
+
+```bash
+python scripts/migrate_json_to_sql.py
+```
+
 ### 1. Local Development Machine
 
 **Best for:** Testing and personal use

env.example

@@ -32,6 +32,11 @@ TIMELINE_HOST=0.0.0.0
 TIMELINE_PORT=8080
 TIMELINE_API_URL=http://127.0.0.1:8080
 TIMELINE_USER_ID=default
+# Single source of truth for timeline + A2A persistence.
+# Unset -> sqlite:///~/.xmcp/xmcp.db (local default)
+# Railway/Postgres -> postgresql://... (or postgres://...; auto-normalized)
+DATABASE_URL=
+# Legacy JSON paths only used by scripts/migrate_json_to_sql.py:
 TIMELINE_STORE_PATH=~/.xmcp/timeline_store.json
 A2A_STORE_PATH=~/.xmcp/a2a_store.json
 TIMELINE_ACTION_AGENT=mcp-orchestrator

requirements.txt

@@ -7,6 +7,8 @@ python-dotenv>=1.2.2
 fastapi>=0.141.1
 uvicorn>=0.52.0
 pydantic>=2.13.4
+sqlalchemy>=2.0.43
+psycopg[binary]>=3.2.10
 
 # X API listener
 tweepy>=4.17.0