Add Substack API client

Author: dynabe

.env.example

@@ -0,0 +1,2 @@
+# Your Cotlet API key. Get it at https://cotlet.dev/app/signup?app=substack
+COTLET_API_KEY=cotlet_...

.gitignore

@@ -0,0 +1,8 @@
+.env
+__pycache__/
+*.pyc
+.venv/
+*.csv
+*.json
+!.github/**/*.json
+!mcp.json

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Take Group sp. z o.o. (Cotlet)
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

assets/logo-white.svg

@@ -0,0 +1,171 @@
+"""Tiny client for the Substack API hosted by Cotlet (https://cotlet.dev).
+
+Every script in this repo imports it. Standard library only, Python 3.9+.
+
+    from cotlet_client import call
+    drafts = call("list_posts", {"status": "draft"})["posts"]
+
+Set COTLET_API_KEY in your environment or in a .env file next to the scripts.
+Get a key at https://cotlet.dev/app/signup?app=substack
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+import os
+import sys
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Any, Iterator
+
+API_BASE = os.environ.get("COTLET_API_BASE", "https://cotlet.dev/api/v1")
+APP = "substack"
+SIGNUP_URL = "https://cotlet.dev/app/signup?app=substack"
+DOCS_URL = "https://cotlet.dev/integrations/substack-api-mcp-server"
+USER_AGENT = "cotletdev-substack-api-mcp-server/1.0 (+https://github.com/cotletdev/substack-api-mcp-server)"
+
+
+class CotletError(Exception):
+    """An error returned by the API, with the code and the link to act on."""
+
+    def __init__(self, status: int, body: dict[str, Any]):
+        self.status = status
+        self.code = body.get("error", "unknown_error")
+        self.body = body
+        link = body.get("connect_url") or body.get("request_access_url") or body.get("docs_url")
+        message = body.get("message", "The request failed.")
+        super().__init__(f"{self.code}: {message}" + (f"\n  -> {link}" if link else ""))
+
+
+def _load_dotenv() -> None:
+    """Read KEY=value lines from .env next to this file, without overriding the environment."""
+    path = Path(__file__).with_name(".env")
+    if not path.exists():
+        return
+    for line in path.read_text().splitlines():
+        line = line.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        key, value = line.split("=", 1)
+        os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))
+
+
+def api_key() -> str:
+    _load_dotenv()
+    key = os.environ.get("COTLET_API_KEY", "").strip()
+    if not key:
+        sys.exit(
+            "COTLET_API_KEY is not set.\n"
+            f"1. Request access and get your key: {SIGNUP_URL}\n"
+            "2. Connect your Substack account once in the Cotlet console.\n"
+            "3. export COTLET_API_KEY=cotlet_...  (or put it in .env)"
+        )
+    return key
+
+
+def _request(method: str, url: str, payload: dict[str, Any] | None = None) -> tuple[int, dict[str, Any]]:
+    data = json.dumps(payload or {}).encode() if method == "POST" else None
+    req = urllib.request.Request(url, data=data, method=method)
+    req.add_header("Authorization", f"Bearer {api_key()}")
+    req.add_header("Content-Type", "application/json")
+    req.add_header("Accept", "application/json")
+    req.add_header("User-Agent", USER_AGENT)
+    for attempt in range(5):
+        try:
+            with urllib.request.urlopen(req, timeout=120) as resp:
+                return resp.status, json.loads(resp.read() or b"{}")
+        except urllib.error.HTTPError as err:
+            body = json.loads(err.read() or b"{}")
+            if err.code == 429 and attempt < 4:
+                time.sleep(int(err.headers.get("Retry-After", "5")))
+                continue
+            raise CotletError(err.code, body) from None
+    raise CotletError(429, {"error": "rate_limited", "message": "Still rate limited after 5 tries."})
+
+
+def wait_for_call(call_id: str, confirm_url: str, timeout: int = 900) -> dict[str, Any]:
+    """Write actions wait for your OK by default. Poll until you approve or decline."""
+    print(f"Waiting for your OK: {confirm_url}", file=sys.stderr)
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        time.sleep(5)
+        _, body = _request("GET", f"{API_BASE}/calls/{call_id}")
+        if body.get("status") not in ("awaiting_confirmation", "running", "queued"):
+            return body.get("result", body)
+    raise TimeoutError(f"No confirmation within {timeout}s for call {call_id}.")
+
+
+def call(endpoint: str, params: dict[str, Any] | None = None, wait: bool = True) -> dict[str, Any]:
+    """Call one Substack endpoint, e.g. call("list_posts", {"status": "draft"})."""
+    clean = {k: v for k, v in (params or {}).items() if v is not None}
+    status, body = _request("POST", f"{API_BASE}/{APP}/{endpoint}", clean)
+    if status == 202 and body.get("status") == "awaiting_confirmation":
+        return wait_for_call(body["call_id"], body["confirm_url"]) if wait else body
+    return body
+
+
+def paginate(endpoint: str, params: dict[str, Any], key: str, max_items: int | None = None) -> Iterator[dict[str, Any]]:
+    """Yield every item of a list endpoint, following next_cursor."""
+    params = dict(params)
+    seen = 0
+    while True:
+        page = call(endpoint, params)
+        for item in page.get(key, []):
+            yield item
+            seen += 1
+            if max_items and seen >= max_items:
+                return
+        cursor = page.get("next_cursor")
+        if not cursor:
+            return
+        params["cursor"] = cursor
+
+
+def write_rows(rows: list[dict[str, Any]], out: str | None) -> None:
+    """Print JSON to stdout, or save .csv / .json when a path is given."""
+    if not out:
+        print(json.dumps(rows, indent=2, ensure_ascii=False))
+        return
+    path = Path(out)
+    if path.suffix.lower() == ".csv":
+        fields: list[str] = []
+        for row in rows:
+            fields += [k for k in row if k not in fields]
+        with path.open("w", newline="", encoding="utf-8") as fh:
+            writer = csv.DictWriter(fh, fieldnames=fields)
+            writer.writeheader()
+            for row in rows:
+                writer.writerow({k: json.dumps(v, ensure_ascii=False) if isinstance(v, (dict, list)) else v for k, v in row.items()})
+    else:
+        path.write_text(json.dumps(rows, indent=2, ensure_ascii=False), encoding="utf-8")
+    print(f"Saved {len(rows)} rows to {path}", file=sys.stderr)
+
+
+def read_markdown(path: str) -> tuple[str, str | None, str]:
+    """Split a .md file into title (first # heading), optional subtitle (a ## or _italic_ line right after it) and body."""
+    lines = Path(path).read_text(encoding="utf-8").splitlines()
+    title, subtitle, start = Path(path).stem.replace("-", " ").capitalize(), None, 0
+    for i, line in enumerate(lines):
+        if line.strip():
+            if line.startswith("# "):
+                title, start = line[2:].strip(), i + 1
+            break
+    rest = lines[start:]
+    while rest and not rest[0].strip():
+        rest.pop(0)
+    if rest and (rest[0].startswith("## ") or (rest[0].startswith("_") and rest[0].rstrip().endswith("_"))):
+        subtitle = rest.pop(0).lstrip("#").strip().strip("_").strip()
+    return title, subtitle, "\n".join(rest).strip() + "\n"
+
+
+def run(main) -> None:
+    """Run a script's main() and turn API errors into a readable message."""
+    try:
+        main()
+    except CotletError as err:
+        sys.exit(f"Error {err.status} {err}")
+    except KeyboardInterrupt:
+        sys.exit(130)