feat: add Directus blog post publisher and MCP tool

- app/services/directus.py: Directus CMS client using docker run
  curlimages/curl on website_cf-internal network; supports static admin
  token with fresh JWT login fallback; get/publish/update blog_posts
- app/api/endpoints/blog.py: POST /api/v1/blog (publish), GET /slug,
  PATCH /id endpoints
- app/api/routes.py: register blog router
- app/core/config.py: add directus_url/token/email/password/network settings
- mcp/server.js: add publish_blog_post and get_blog_post MCP tools

Key gotcha: Directus filter[field][_eq] brackets must be percent-encoded
when passed as a curl CLI URL arg — raw brackets cause curl to exit
non-zero with empty stderr.

app/api/endpoints/blog.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+
+from app.services.directus import get_blog_post, publish_blog_post, update_blog_post
+
+router = APIRouter(prefix="/blog", tags=["blog"])
+
+
+class PublishRequest(BaseModel):
+    title: str
+    body: str
+    slug: str | None = None
+    tags: list[str] | None = None
+    author: str | None = None
+    seo_description: str | None = None
+    published_at: str | None = None  # ISO 8601; defaults to now
+
+
+class UpdateRequest(BaseModel):
+    title: str | None = None
+    body: str | None = None
+    tags: list[str] | None = None
+    seo_description: str | None = None
+    published_at: str | None = None
+
+
+@router.post("", summary="Publish a blog post to the CircuitForge website via Directus")
+def publish(req: PublishRequest) -> dict:
+    try:
+        item = publish_blog_post(
+            title=req.title,
+            body=req.body,
+            slug=req.slug,
+            tags=req.tags,
+            author=req.author,
+            seo_description=req.seo_description,
+            published_at=req.published_at,
+        )
+        return {"ok": True, "post": item}
+    except Exception as e:
+        raise HTTPException(status_code=502, detail=str(e))
+
+
+@router.get("/{slug}", summary="Fetch a blog post by slug")
+def get_post(slug: str) -> dict:
+    post = get_blog_post(slug)
+    if post is None:
+        raise HTTPException(status_code=404, detail=f"No blog post with slug '{slug}'")
+    return post
+
+
+@router.patch("/{post_id}", summary="Update an existing blog post by Directus ID")
+def update(post_id: int, req: UpdateRequest) -> dict:
+    fields = req.model_dump(exclude_none=True)
+    if not fields:
+        raise HTTPException(status_code=400, detail="No fields to update")
+    try:
+        item = update_blog_post(post_id, fields)
+        return {"ok": True, "post": item}
+    except Exception as e:
+        raise HTTPException(status_code=502, detail=str(e))

app/api/routes.py

@@ -1,6 +1,6 @@
 from fastapi import FastAPI
 
-from app.api.endpoints import campaigns, opportunities, posts, scheduler, signals, subs
+from app.api.endpoints import blog, campaigns, opportunities, posts, scheduler, signals, subs
 
 
 def register_routes(app: FastAPI) -> None:
@@ -10,3 +10,4 @@ def register_routes(app: FastAPI) -> None:
     app.include_router(scheduler.router, prefix="/api/v1")
     app.include_router(opportunities.router, prefix="/api/v1")
     app.include_router(signals.router, prefix="/api/v1")
+    app.include_router(blog.router, prefix="/api/v1")

app/core/config.py

@@ -21,6 +21,13 @@ class Settings(BaseSettings):
     # Scheduler
     scheduler_enabled: bool = True
 
+    # Directus (CircuitForge website CMS)
+    directus_url: str = "http://172.31.0.4:8055"
+    directus_admin_token: str = ""
+    directus_admin_email: str = ""
+    directus_admin_password: str = ""
+    directus_network: str = "website_cf-internal"
+
     # Signal scraper
     scraper_enabled: bool = True
     scraper_interval_mins: int = 30          # how often to poll (per full pass of all subs)

app/services/directus.py

@@ -0,0 +1,148 @@
+"""
+Directus blog post publisher for the CircuitForge website CMS.
+
+Directus runs in Docker on the website_cf-internal network and is not
+directly reachable from host processes. We shell out to a one-shot
+curlimages/curl container joined to that network.
+
+Collection: blog_posts
+Fields: id, title, slug, body, published_at, tags, author, seo_description
+
+Environment variables (via Magpie .env):
+    DIRECTUS_URL        Base URL inside the cf-internal network
+                        (default: http://172.31.0.4:8055)
+    DIRECTUS_ADMIN_TOKEN  Static admin token
+    DIRECTUS_ADMIN_EMAIL  Admin email (for fresh JWT fallback)
+    DIRECTUS_ADMIN_PASSWORD
+    DIRECTUS_NETWORK    Docker network name
+                        (default: website_cf-internal)
+
+IP gotcha: 172.31.0.4 is the current cf-directus address on website_cf-internal.
+If calls start returning connection errors run:
+    docker inspect cf-directus --format '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}'
+and update DIRECTUS_URL in Magpie's .env.
+"""
+from __future__ import annotations
+
+import json
+import re
+import subprocess
+from datetime import datetime, timezone
+from typing import Any
+
+from app.core.config import get_settings
+
+_CURL_IMAGE = "curlimages/curl:latest"
+
+
+def _curl(method: str, path: str, token: str, body: dict[str, Any] | None = None) -> dict:
+    """Run a curl request inside a container on the cf-internal network."""
+    cfg = get_settings()
+    url = f"{cfg.directus_url}{path}"
+    cmd = [
+        "docker", "run", "--rm",
+        "--network", cfg.directus_network,
+        _CURL_IMAGE,
+        "-sf", "-X", method, url,
+        "-H", f"Authorization: Bearer {token}",
+        "-H", "Content-Type: application/json",
+    ]
+    if body is not None:
+        cmd += ["--data", json.dumps(body)]
+
+    result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
+    if result.returncode != 0:
+        raise RuntimeError(f"Directus {method} {path} failed: {result.stderr.strip()}")
+    if not result.stdout.strip():
+        return {}
+    return json.loads(result.stdout)
+
+
+def _get_token() -> str:
+    """Return a usable token: static admin token, or fresh JWT via login."""
+    cfg = get_settings()
+    if cfg.directus_admin_token:
+        return cfg.directus_admin_token
+    if not (cfg.directus_admin_email and cfg.directus_admin_password):
+        raise RuntimeError(
+            "No Directus credentials configured. "
+            "Set DIRECTUS_ADMIN_TOKEN or DIRECTUS_ADMIN_EMAIL + DIRECTUS_ADMIN_PASSWORD."
+        )
+    resp = _curl("POST", "/auth/login", token="", body={
+        "email": cfg.directus_admin_email,
+        "password": cfg.directus_admin_password,
+    })
+    access_token = resp.get("data", {}).get("access_token")
+    if not access_token:
+        raise RuntimeError(f"Directus login failed: {resp}")
+    return access_token
+
+
+def slugify(text: str) -> str:
+    slug = text.lower().strip()
+    slug = re.sub(r"[^\w\s-]", "", slug)
+    slug = re.sub(r"[\s_]+", "-", slug)
+    slug = re.sub(r"-+", "-", slug).strip("-")
+    return slug
+
+
+def publish_blog_post(
+    title: str,
+    body: str,
+    slug: str | None = None,
+    tags: list[str] | None = None,
+    author: str | None = None,
+    seo_description: str | None = None,
+    published_at: str | None = None,
+) -> dict:
+    """
+    Create and publish a blog post in Directus.
+
+    Returns the created item dict (id, slug, title, ...).
+
+    published_at defaults to now (UTC ISO 8601). Pass None or omit to publish
+    immediately. Pass a future timestamp to schedule.
+    """
+    token = _get_token()
+
+    _slug = slug or slugify(title)
+    _published_at = published_at or datetime.now(timezone.utc).isoformat()
+
+    payload: dict[str, Any] = {
+        "title": title,
+        "slug": _slug,
+        "body": body,
+        "published_at": _published_at,
+    }
+    if tags:
+        payload["tags"] = tags
+    if author:
+        payload["author"] = author
+    if seo_description:
+        payload["seo_description"] = seo_description
+
+    resp = _curl("POST", "/items/blog_posts", token=token, body=payload)
+    item = resp.get("data", resp)
+    return item
+
+
+def get_blog_post(slug: str) -> dict | None:
+    """Fetch a blog post by slug. Returns None if not found."""
+    from urllib.parse import quote
+    token = _get_token()
+    # Directus filter syntax uses brackets which must be percent-encoded for curl CLI
+    filter_param = f"filter%5Bslug%5D%5B_eq%5D={quote(slug, safe='')}"
+    resp = _curl(
+        "GET",
+        f"/items/blog_posts?{filter_param}&limit=1",
+        token=token,
+    )
+    items = resp.get("data", [])
+    return items[0] if items else None
+
+
+def update_blog_post(post_id: int, fields: dict[str, Any]) -> dict:
+    """Patch an existing blog post by ID."""
+    token = _get_token()
+    resp = _curl("PATCH", f"/items/blog_posts/{post_id}", token=token, body=fields)
+    return resp.get("data", resp)

mcp/server.js

@@ -308,6 +308,36 @@ const TOOLS = [
     },
   },
 
+  // Blog (Directus)
+  {
+    name: 'publish_blog_post',
+    description: 'Publish a blog post to the CircuitForge website via Directus. Defaults to published immediately. Returns the created post including its id and slug.',
+    inputSchema: {
+      type: 'object',
+      properties: {
+        title: { type: 'string', description: 'Post title' },
+        body: { type: 'string', description: 'Post body (Markdown)' },
+        slug: { type: 'string', description: 'URL slug — auto-generated from title if omitted' },
+        tags: { type: 'array', items: { type: 'string' }, description: 'Tag list (e.g. ["sprint-review", "kiwi"])' },
+        author: { type: 'string', description: 'Author name (optional)' },
+        seo_description: { type: 'string', description: 'Short SEO/meta description (optional)' },
+        published_at: { type: 'string', description: 'ISO 8601 publish timestamp — defaults to now' },
+      },
+      required: ['title', 'body'],
+    },
+  },
+  {
+    name: 'get_blog_post',
+    description: 'Fetch an existing blog post by its URL slug.',
+    inputSchema: {
+      type: 'object',
+      properties: {
+        slug: { type: 'string', description: 'The post slug (e.g. "2026-04-28-sprint-review")' },
+      },
+      required: ['slug'],
+    },
+  },
+
   // Scheduler
   {
     name: 'scheduler_status',
@@ -382,6 +412,13 @@ async function callTool(name, args) {
       return await api('PUT', `/subs/${sub}`, body);
     }
 
+    case 'publish_blog_post': {
+      const { title, body, ...rest } = args;
+      return await api('POST', '/blog', { title, body, ...rest });
+    }
+    case 'get_blog_post':
+      return await api('GET', `/blog/${encodeURIComponent(args.slug)}`);
+
     case 'scheduler_status':
       return await api('GET', '/scheduler/status');