Build a User-Scoped MCP Server on macOS

mkdir -p ~/projects && cd ~/projects
uv init --bare --name chirp-sidecar \
  --description "A user-scoped MCP sidecar for one Chirp account" \
  user-scoped-mcp-server-macos
cd user-scoped-mcp-server-macos
touch .gitignore

uv add fastmcp httpx mcp
uv add --dev pytest pytest-asyncio

[project]
name = "chirp-sidecar"
version = "0.1.0"
description = "A user-scoped MCP sidecar for one Chirp account"
requires-python = ">=3.12"
dependencies = [
    "fastmcp>=4.0.2",
    "httpx>=0.28.1",
    "mcp>=2.1.1",
]

[dependency-groups]
dev = [
    "pytest>=9.1.1",
    "pytest-asyncio>=1.4.0",
]

touch chirpd.py

"""Chirp: a small social service that exists so this project has one to talk to.

Chirp is not real. It is a single-file stand-in for the kind of service a
personal MCP sidecar usually fronts: it has accounts, it issues a credential per
account, and it refuses to let one account write as another. That is every
property the sidecar needs and nothing else, which is the point.

Four endpoints:

    POST /accounts              create an account, get its app password once
    POST /session               username + app password -> a bearer token
    POST /posts                 write a post as the token's account
    GET  /users/<username>      public profile, no credential needed

State lives in `chirp-data.json` next to this file, so an account you create
survives a restart. Run it with `make serve-chirp`, or:

    uv run python chirpd.py --port 8787
"""

from __future__ import annotations

import argparse
import json
import secrets
import threading
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlparse

DATA_FILE = Path(__file__).resolve().parent / "chirp-data.json"
MAX_POST_CHARS = 280

_LOCK = threading.Lock()


def load() -> dict:
    """Read the whole store. Missing or corrupt file means an empty service."""
    try:
        return json.loads(DATA_FILE.read_text())
    except (FileNotFoundError, ValueError):
        return {"accounts": {}, "posts": []}


def save(store: dict) -> None:
    DATA_FILE.write_text(json.dumps(store, indent=2) + "\n")


def make_app_password() -> str:
    """Four groups of four, the shape most services use for a scoped credential."""
    alphabet = "abcdefghijkmnpqrstuvwxyz23456789"
    raw = "".join(secrets.choice(alphabet) for _ in range(16))
    return "-".join(raw[i : i + 4] for i in range(0, 16, 4))


class ChirpHandler(BaseHTTPRequestHandler):
    """One method per route. No framework, so the whole service reads top to bottom."""

    server_version = "chirpd/1.0"

    def do_POST(self) -> None:  # noqa: N802 - name fixed by BaseHTTPRequestHandler
        path = urlparse(self.path).path
        body = self._body()
        if path == "/accounts":
            return self._create_account(body)
        if path == "/session":
            return self._create_session(body)
        if path == "/posts":
            return self._create_post(body)
        self._error(404, "not_found", f"No such route: {path}")

    def do_GET(self) -> None:  # noqa: N802 - name fixed by BaseHTTPRequestHandler
        path = urlparse(self.path).path
        if path.startswith("/users/"):
            return self._get_user(path.removeprefix("/users/").strip("/"))
        self._error(404, "not_found", f"No such route: {path}")

    def _create_account(self, body: dict) -> None:
        username = str(body.get("username") or "").strip().lower()
        if not username.isalnum():
            return self._error(400, "bad_username", "Username must be letters and digits only.")
        with _LOCK:
            store = load()
            if username in store["accounts"]:
                return self._error(409, "taken", f"Account {username!r} already exists.")
            app_password = make_app_password()
            store["accounts"][username] = {
                "user_id": f"chirp_{secrets.token_hex(6)}",
                "app_password": app_password,
                "display_name": str(body.get("display_name") or username),
                "created_at": _now(),
            }
            save(store)
        account = load()["accounts"][username]
        # The app password is returned once, here, and never again.
        self._json(201, {"username": username, "user_id": account["user_id"], "app_password": app_password})

    def _create_session(self, body: dict) -> None:
        username = str(body.get("username") or "").strip().lower()
        password = str(body.get("app_password") or "")
        account = load()["accounts"].get(username)
        if account is None or not secrets.compare_digest(
            password.encode("utf-8"), account["app_password"].encode("utf-8")
        ):
            return self._error(401, "bad_credentials", "Unknown username or wrong app password.")
        self._json(
            200,
            {
                "user_id": account["user_id"],
                "username": username,
                "token": f"chirp-token.{account['user_id']}",
            },
        )

    def _create_post(self, body: dict) -> None:
        token = (self.headers.get("Authorization") or "").removeprefix("Bearer ").strip()
        if not token.startswith("chirp-token."):
            return self._error(401, "no_token", "Missing or malformed bearer token.")
        user_id = token.removeprefix("chirp-token.")

        with _LOCK:
            store = load()
            owner = next((u for u, a in store["accounts"].items() if a["user_id"] == user_id), None)
            if owner is None:
                return self._error(401, "no_token", "Token does not belong to any account.")
            text = str(body.get("text") or "")
            if not text.strip():
                return self._error(400, "empty", "A post needs text.")
            if len(text) > MAX_POST_CHARS:
                return self._error(400, "too_long", f"{len(text)} characters; the limit is {MAX_POST_CHARS}.")
            post = {
                "post_id": f"post_{secrets.token_hex(4)}",
                "author": owner,
                "author_id": user_id,
                "text": text,
                "created_at": _now(),
            }
            store["posts"].append(post)
            save(store)
        self._json(201, post | {"url": f"http://{self.headers.get('Host', 'localhost')}/users/{owner}"})

    def _get_user(self, username: str) -> None:
        store = load()
        account = store["accounts"].get(username.lower())
        if account is None:
            return self._error(404, "no_such_user", f"No account named {username!r}.")
        posts = [p for p in store["posts"] if p["author"] == username.lower()]
        self._json(
            200,
            {
                "username": username.lower(),
                "user_id": account["user_id"],
                "display_name": account["display_name"],
                "post_count": len(posts),
                "posts": posts[-10:],
            },
        )

    def _body(self) -> dict:
        length = int(self.headers.get("Content-Length") or 0)
        try:
            return json.loads(self.rfile.read(length) or b"{}") if length else {}
        except ValueError:
            return {}

    def _json(self, status: int, payload: object) -> None:
        raw = json.dumps(payload).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(raw)))
        self.end_headers()
        self.wfile.write(raw)

    def _error(self, status: int, code: str, message: str) -> None:
        self._json(status, {"error": code, "message": message})

    def log_message(self, fmt: str, *args) -> None:
        """Silence the default access log; the demos own the terminal."""


def _now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")


def start_in_thread(port: int = 0) -> tuple[ThreadingHTTPServer, str]:
    """Serve on a daemon thread and return the server plus its base URL."""
    server = ThreadingHTTPServer(("127.0.0.1", port), ChirpHandler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server, f"http://127.0.0.1:{server.server_address[1]}"


def main() -> None:
    parser = argparse.ArgumentParser(description="Run the Chirp service.")
    parser.add_argument("--port", type=int, default=8787)
    args = parser.parse_args()
    server, url = start_in_thread(args.port)
    print(f"chirpd listening on {url} (Ctrl-C to stop)", flush=True)
    print(f"data file: {DATA_FILE}", flush=True)
    try:
        threading.Event().wait()
    except KeyboardInterrupt:
        print("\nchirpd stopped.")
    finally:
        server.shutdown()
        server.server_close()


if __name__ == "__main__":
    main()

uv run python chirpd.py --port 8787

curl -s -X POST http://127.0.0.1:8787/accounts \
  -H 'content-type: application/json' \
  -d '{"username":"alice"}' | python3 -m json.tool

{
    "username": "alice",
    "user_id": "chirp_f07d78077a33",
    "app_password": "dg3r-5m39-sm36-987j"
}

CHIRP_USERNAME=alice
CHIRP_APP_PASSWORD=dg3r-5m39-sm36-987j

touch identity.py

"""The scope boundary: who this sidecar acts as.

Everything account-specific enters the process here and nowhere else. The rest
of the project is generic; run two copies with two different environments and
you get two servers that cannot touch each other's account.

`AccountConfig.from_env()` is deliberately strict. A sidecar with no username
must fail at startup, not on the first tool call, because a server that starts
successfully is advertised to the MCP client as working.
"""

from __future__ import annotations

import os
from dataclasses import dataclass
from urllib.parse import urlparse

DEFAULT_CHIRP_URL = "http://127.0.0.1:8787"

# Hosts allowed to receive a credential over plaintext HTTP.
PLAINTEXT_OK_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"})


class ConfigError(RuntimeError):
    """The environment does not describe a usable account."""


@dataclass(frozen=True)
class AccountConfig:
    """The one account this process is allowed to act as."""

    username: str
    app_password: str
    chirp_url: str = DEFAULT_CHIRP_URL
    allow_publish: bool = False

    @classmethod
    def from_env(cls, env: dict[str, str] | None = None) -> AccountConfig:
        """Build the config from environment variables, or raise `ConfigError`."""
        env = os.environ if env is None else env

        username = env.get("CHIRP_USERNAME", "").strip().lstrip("@").lower()
        app_password = env.get("CHIRP_APP_PASSWORD", "").strip()
        chirp_url = env.get("CHIRP_URL", DEFAULT_CHIRP_URL).strip().rstrip("/")

        missing = [
            name
            for name, value in (("CHIRP_USERNAME", username), ("CHIRP_APP_PASSWORD", app_password))
            if not value
        ]
        if missing:
            raise ConfigError(
                "This server is user-scoped and has no account to act as. "
                f"Set {' and '.join(missing)} in the MCP client's env block."
            )

        if not username.isalnum():
            raise ConfigError(f"CHIRP_USERNAME must be letters and digits only, got {username!r}.")

        parsed = urlparse(chirp_url)
        host, scheme = parsed.hostname or "", parsed.scheme
        if scheme not in {"http", "https"}:
            raise ConfigError(f"CHIRP_URL must be an http(s) URL, got {chirp_url!r}.")
        if scheme == "http" and host not in PLAINTEXT_OK_HOSTS:
            raise ConfigError(
                f"Refusing to send an app password over plaintext HTTP to {host!r}. "
                "Use https, or point CHIRP_URL at a service on this machine."
            )

        return cls(
            username=username,
            app_password=app_password,
            chirp_url=chirp_url,
            allow_publish=_flag(env.get("CHIRP_ALLOW_PUBLISH")),
        )

    def describe(self) -> dict[str, object]:
        """The scope, safe to show a caller. Never includes the app password."""
        return {
            "username": self.username,
            "chirp_url": self.chirp_url,
            "publish_allowed": self.allow_publish,
        }


def _flag(raw: str | None) -> bool:
    """Read a boolean env var. Absent, empty, and `0`/`false`/`no` are all off."""
    return (raw or "").strip().lower() in {"1", "true", "yes", "on"}

touch chirp_client.py

"""An HTTP client for the three Chirp calls the sidecar makes.

Every failure becomes a `ChirpError`, so nothing above this file has to know
what `httpx` raises. The `transport` parameter is the test seam: passing an
`httpx.MockTransport` swaps the whole network layer without patching anything.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

import httpx

DEFAULT_TIMEOUT = 10.0


class ChirpError(RuntimeError):
    """A Chirp call failed. Carries the service's error code when it sent one."""

    def __init__(self, message: str, *, code: str | None = None, status: int | None = None):
        super().__init__(message)
        self.code = code
        self.status = status


@dataclass(frozen=True)
class Session:
    """An authenticated session, bound to exactly one account."""

    user_id: str
    username: str
    token: str


class ChirpClient:
    """Calls against one Chirp host, with a bounded timeout."""

    def __init__(self, base_url: str, *, timeout: float = DEFAULT_TIMEOUT, transport: httpx.AsyncBaseTransport | None = None):
        self._base_url = base_url.rstrip("/")
        self._timeout = timeout
        self._transport = transport

    async def get_profile(self, username: str) -> dict[str, Any]:
        """Read a public profile. Unauthenticated, so it needs no credential."""
        return await self._request("GET", f"/users/{username}")

    async def create_session(self, username: str, app_password: str) -> Session:
        """Exchange a username and app password for a bearer token."""
        body = await self._request("POST", "/session", json={"username": username, "app_password": app_password})
        return Session(user_id=body["user_id"], username=body["username"], token=body["token"])

    async def create_post(self, session: Session, text: str) -> dict[str, Any]:
        """Write one post as the session's account."""
        return await self._request(
            "POST",
            "/posts",
            json={"text": text},
            headers={"Authorization": f"Bearer {session.token}"},
        )

    async def _request(self, method: str, path: str, *, json: dict[str, Any] | None = None, headers: dict[str, str] | None = None) -> dict[str, Any]:
        """One round trip, with every failure turned into a `ChirpError`."""
        url = f"{self._base_url}{path}"
        try:
            async with httpx.AsyncClient(timeout=self._timeout, transport=self._transport) as client:
                response = await client.request(method, url, json=json, headers=headers)
        except httpx.TimeoutException as exc:
            raise ChirpError(f"{path} timed out after {self._timeout:g}s.") from exc
        except httpx.HTTPError as exc:
            raise ChirpError(f"Could not reach Chirp at {self._base_url}: {exc}.") from exc

        if response.status_code >= 400:
            code, message = _error_of(response)
            raise ChirpError(f"{path} failed ({response.status_code} {code}): {message}", code=code, status=response.status_code)
        try:
            return response.json()
        except ValueError as exc:
            raise ChirpError(f"{path} returned a non-JSON body.") from exc


def _error_of(response: httpx.Response) -> tuple[str, str]:
    """Chirp errors are `{"error": code, "message": text}`; tolerate anything else."""
    try:
        body = response.json()
    except ValueError:
        return "unknown", response.text[:200] or response.reason_phrase
    if not isinstance(body, dict):
        return "unknown", str(body)[:200]
    return str(body.get("error") or "unknown"), str(body.get("message") or response.reason_phrase)

touch server.py

"""A user-scoped MCP sidecar for one Chirp account.

The server is built by a factory rather than declared at module scope, because
the whole point of a user-scoped sidecar is that the *code* is generic and the
*instance* is specific. `create_server(config)` takes the account it should act
as; `main()` reads that account from the environment and refuses to start
without one.

Three tools, in increasing order of consequence:

- `whoami` reads the configured identity and needs no credential.
- `draft_post` checks the text against Chirp's limit, entirely offline.
- `publish_post` writes to the account, and only when the environment has
  explicitly allowed writes.
"""

from __future__ import annotations

from typing import Any

from fastmcp import FastMCP
from fastmcp.exceptions import ToolError
from mcp.types import ToolAnnotations

from chirp_client import ChirpClient, ChirpError
from identity import AccountConfig, ConfigError

MAX_POST_CHARS = 280


def create_server(config: AccountConfig, *, client: ChirpClient | None = None) -> FastMCP:
    """Build a server bound to exactly one account."""
    chirp = client or ChirpClient(config.chirp_url)
    mcp = FastMCP(f"chirp-{config.username}", mask_error_details=True)

    @mcp.tool(
        annotations=ToolAnnotations(
            title="Show this server's account",
            read_only_hint=True,
            idempotent_hint=True,
            open_world_hint=True,
        )
    )
    async def whoami() -> dict[str, Any]:
        """Report which Chirp account this server acts as, and whether it may post."""
        scope = config.describe()
        try:
            profile = await chirp.get_profile(config.username)
        except ChirpError as exc:
            raise ToolError(f"Could not read the profile for {config.username}: {exc}") from exc
        scope["user_id"] = profile["user_id"]
        scope["post_count"] = profile["post_count"]
        return scope

    @mcp.tool(
        annotations=ToolAnnotations(
            title="Draft a post without publishing",
            read_only_hint=True,
            idempotent_hint=True,
            open_world_hint=False,
        )
    )
    def draft_post(text: str) -> dict[str, Any]:
        """Check post text against Chirp's length limit and return what would be sent."""
        length = len(text)
        if not text.strip():
            return {"valid": False, "length": length, "reason": "a post needs text"}
        if length > MAX_POST_CHARS:
            return {"valid": False, "length": length, "reason": f"{length} characters, {MAX_POST_CHARS} allowed"}
        return {
            "valid": True,
            "length": length,
            "remaining": MAX_POST_CHARS - length,
            "account": config.username,
            "text": text,
        }

    @mcp.tool(
        annotations=ToolAnnotations(
            title="Publish a post to this account",
            read_only_hint=False,
            destructive_hint=False,
            idempotent_hint=False,
            open_world_hint=True,
        )
    )
    async def publish_post(text: str) -> dict[str, Any]:
        """Publish `text` as a post on this server's account."""
        if not config.allow_publish:
            raise ToolError(
                "Publishing is disabled for this server. Set CHIRP_ALLOW_PUBLISH=1 "
                "in its env block to enable it."
            )
        draft = draft_post(text)
        if not draft["valid"]:
            raise ToolError(f"Not published: {draft['reason']}.")

        try:
            session = await chirp.create_session(config.username, config.app_password)
            if session.username != config.username:
                raise ToolError(
                    f"Refusing to write: session belongs to {session.username}, "
                    f"not the configured {config.username}."
                )
            result = await chirp.create_post(session, text)
        except ChirpError as exc:
            raise ToolError(f"Publish failed: {exc}") from exc

        return {
            "post_id": result["post_id"],
            "account": result["author"],
            "user_id": result["author_id"],
            "url": result.get("url", ""),
        }

    return mcp


def main() -> None:
    """Run the sidecar over stdio, or exit with a readable configuration error."""
    try:
        config = AccountConfig.from_env()
    except ConfigError as exc:
        raise SystemExit(f"chirp-sidecar: {exc}")
    create_server(config).run()


if __name__ == "__main__":
    main()

touch demo.py

"""Drive one sidecar instance and print what its scope allows.

Reads the account from the environment exactly as `server.py` does, so the
output is a readout of this shell's environment and nothing else.

    uv run python demo.py
"""

import asyncio
import json
import logging

from fastmcp import Client

from identity import AccountConfig, ConfigError
from server import create_server

# A deliberately refused tool call is not a server fault; keep it out of stderr
# so this demo prints only its own output.
logging.getLogger("fastmcp").setLevel(logging.CRITICAL)


async def report(config: AccountConfig) -> None:
    """Call every tool once and print the result, refusals included."""
    print(f"== instance for {config.username}  (chirp {config.chirp_url}, publish={config.allow_publish})")
    async with Client(create_server(config)) as client:
        for tool in await client.list_tools():
            print(f"   tool {tool.name:13} read_only={tool.annotations.read_only_hint}")

        res = await client.call_tool("whoami", {}, raise_on_error=False)
        if res.is_error:
            print("\n   whoami       failed:", res.content[0].text)
            return
        print("\n   whoami       ", json.dumps(res.data, sort_keys=True))

        res = await client.call_tool("draft_post", {"text": "Shipped a user-scoped sidecar."})
        print("   draft_post   ", json.dumps(res.data, sort_keys=True))

        res = await client.call_tool("draft_post", {"text": "x" * 281})
        print("   draft_post   valid =", res.data["valid"], "reason =", res.data["reason"])

        text = f"Hello from {config.username}, posted by a user-scoped MCP sidecar."
        res = await client.call_tool("publish_post", {"text": text}, raise_on_error=False)
        if res.is_error:
            print("   publish_post refused:", res.content[0].text)
        else:
            print("   publish_post ", json.dumps(res.data, sort_keys=True))


def main() -> None:
    try:
        config = AccountConfig.from_env()
    except ConfigError as exc:
        raise SystemExit(f"demo: {exc}")
    asyncio.run(report(config))


if __name__ == "__main__":
    main()

export CHIRP_USERNAME=alice
export CHIRP_APP_PASSWORD=dg3r-5m39-sm36-987j
uv run python demo.py

export CHIRP_ALLOW_PUBLISH=1
uv run python demo.py

curl -s http://127.0.0.1:8787/users/alice | python3 -m json.tool

{
    "username": "alice",
    "user_id": "chirp_f07d78077a33",
    "display_name": "alice",
    "post_count": 1,
    "posts": [
        {
            "post_id": "post_b331039a",
            "author": "alice",
            "author_id": "chirp_f07d78077a33",
            "text": "Hello from alice, posted by a user-scoped MCP sidecar.",
            "created_at": "2026-09-04T14:07:56Z"
        }
    ]
}

touch swap.py

"""The scoping proof: one codebase, two environments, two accounts.

Nothing here reaches into the server's internals. It creates two Chirp accounts,
builds two configs from two dictionaries that stand in for two MCP client `env`
blocks, hands each to the same `create_server`, and then asks Chirp itself what
it stored.
"""

import asyncio
import logging

import httpx
from fastmcp import Client

import chirpd
from identity import AccountConfig
from server import create_server

logging.getLogger("fastmcp").setLevel(logging.CRITICAL)

USERNAMES = ("alice", "bob")


async def signup(url: str, username: str) -> str:
    """Create an account on the running service and return its app password."""
    async with httpx.AsyncClient(timeout=5) as client:
        response = await client.post(f"{url}/accounts", json={"username": username})
    return response.json()["app_password"]


async def publish_as(username: str, app_password: str, url: str) -> dict:
    """Start a sidecar scoped to `username` and publish one post from it."""
    config = AccountConfig.from_env(
        {
            "CHIRP_USERNAME": username,
            "CHIRP_APP_PASSWORD": app_password,
            "CHIRP_URL": url,
            "CHIRP_ALLOW_PUBLISH": "1",
        }
    )
    async with Client(create_server(config)) as client:
        return (await client.call_tool("publish_post", {"text": f"hello from {username}"})).data


async def profile(url: str, username: str) -> dict:
    async with httpx.AsyncClient(timeout=5) as client:
        return (await client.get(f"{url}/users/{username}")).json()


async def main() -> None:
    # A throwaway service on its own port, so this never touches your real data file.
    original, chirpd.DATA_FILE = chirpd.DATA_FILE, chirpd.DATA_FILE.with_name("chirp-swap-data.json")
    chirpd.DATA_FILE.unlink(missing_ok=True)
    server, url = chirpd.start_in_thread()
    try:
        for username in USERNAMES:
            password = await signup(url, username)
            result = await publish_as(username, password, url)
            print(f"{username:8} -> post {result['post_id']} as {result['user_id']}")

        print("\nWhat Chirp actually stored:")
        for username in USERNAMES:
            data = await profile(url, username)
            texts = [p["text"] for p in data["posts"]]
            print(f"  {username:8} {data['user_id']}  {texts}")

        ids = {(await profile(url, u))["user_id"] for u in USERNAMES}
        print(f"\nDistinct accounts written: {len(ids)}")
    finally:
        server.shutdown()
        server.server_close()
        chirpd.DATA_FILE.unlink(missing_ok=True)
        chirpd.DATA_FILE = original


if __name__ == "__main__":
    asyncio.run(main())

uv run python swap.py

mkdir -p tests
touch pytest.ini tests/__init__.py tests/test_identity.py tests/test_server.py tests/test_chirpd.py

[pytest]
asyncio_mode = auto
filterwarnings =
    ignore::DeprecationWarning

"""The scope boundary is the thing most worth testing: it is the security claim."""

import pytest

from identity import DEFAULT_CHIRP_URL, AccountConfig, ConfigError

GOOD = {"CHIRP_USERNAME": "alice", "CHIRP_APP_PASSWORD": "abcd-efgh-ijkl-mnpq"}


def test_defaults_to_the_local_service_and_no_publishing():
    config = AccountConfig.from_env(GOOD)
    assert config.chirp_url == DEFAULT_CHIRP_URL
    assert config.allow_publish is False


@pytest.mark.parametrize(
    "raw,expected",
    [("1", True), ("true", True), ("YES", True), ("0", False), ("", False), (None, False)],
)
def test_publish_flag_is_opt_in(raw, expected):
    env = dict(GOOD) | ({"CHIRP_ALLOW_PUBLISH": raw} if raw is not None else {})
    assert AccountConfig.from_env(env).allow_publish is expected


def test_username_is_normalised():
    assert AccountConfig.from_env(dict(GOOD, CHIRP_USERNAME="@Alice")).username == "alice"


@pytest.mark.parametrize("missing", ["CHIRP_USERNAME", "CHIRP_APP_PASSWORD"])
def test_missing_credentials_name_themselves(missing):
    env = {k: v for k, v in GOOD.items() if k != missing}
    with pytest.raises(ConfigError, match=missing):
        AccountConfig.from_env(env)


def test_a_username_with_punctuation_is_rejected():
    with pytest.raises(ConfigError, match="letters and digits"):
        AccountConfig.from_env(dict(GOOD, CHIRP_USERNAME="alice.smith"))


def test_plaintext_http_to_a_remote_host_is_refused():
    with pytest.raises(ConfigError, match="plaintext HTTP"):
        AccountConfig.from_env(dict(GOOD, CHIRP_URL="http://chirp.example.com"))


def test_plaintext_http_to_localhost_is_allowed():
    config = AccountConfig.from_env(dict(GOOD, CHIRP_URL="http://127.0.0.1:8787/"))
    assert config.chirp_url == "http://127.0.0.1:8787"


def test_describe_never_leaks_the_app_password():
    assert "abcd-efgh-ijkl-mnpq" not in repr(AccountConfig.from_env(GOOD).describe())

"""Tool behaviour against a mock Chirp, so no test needs the service running.

`httpx.MockTransport` is injected into the client, which means these tests cover
the requests the sidecar actually sends, including the guarantee that a post is
written with the token the session returned.
"""

import json

import httpx
import pytest
from fastmcp import Client

from chirp_client import ChirpClient
from identity import AccountConfig
from server import MAX_POST_CHARS, create_server

IDS = {"alice": "chirp_alice0000", "bob": "chirp_bob00000"}


def config_for(username: str, **overrides) -> AccountConfig:
    return AccountConfig.from_env(
        {"CHIRP_USERNAME": username, "CHIRP_APP_PASSWORD": "abcd-efgh-ijkl-mnpq", **overrides}
    )


def fake_chirp(seen: list[httpx.Request], *, session_username: str | None = None) -> ChirpClient:
    """A Chirp that logs every request and answers the three calls we make."""

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(request)
        path = request.url.path
        if path.startswith("/users/"):
            username = path.removeprefix("/users/")
            return httpx.Response(200, json={"username": username, "user_id": IDS[username], "display_name": username, "post_count": 3, "posts": []})
        if path == "/session":
            username = json.loads(request.content)["username"]
            returned = session_username or username
            return httpx.Response(200, json={"user_id": IDS[username], "username": returned, "token": f"chirp-token.{IDS[username]}"})
        if path == "/posts":
            user_id = request.headers["Authorization"].removeprefix("Bearer chirp-token.")
            author = next(u for u, i in IDS.items() if i == user_id)
            return httpx.Response(201, json={"post_id": "post_0001", "author": author, "author_id": user_id, "text": json.loads(request.content)["text"], "url": f"http://chirp.test/users/{author}"})
        return httpx.Response(404, json={"error": "not_found", "message": path})

    return ChirpClient("http://chirp.test", transport=httpx.MockTransport(handler))


async def test_whoami_reports_the_scope_and_the_account_id():
    server = create_server(config_for("alice"), client=fake_chirp([]))
    async with Client(server) as client:
        data = (await client.call_tool("whoami", {})).data
    assert data == {
        "username": "alice",
        "chirp_url": "http://127.0.0.1:8787",
        "publish_allowed": False,
        "user_id": IDS["alice"],
        "post_count": 3,
    }


async def test_draft_post_makes_no_network_calls():
    seen: list[httpx.Request] = []
    server = create_server(config_for("alice"), client=fake_chirp(seen))
    async with Client(server) as client:
        data = (await client.call_tool("draft_post", {"text": "hello"})).data
    assert data["valid"] is True
    assert data["remaining"] == MAX_POST_CHARS - 5
    assert seen == []


@pytest.mark.parametrize(
    "text,reason",
    [("x" * (MAX_POST_CHARS + 1), "281 characters, 280 allowed"), ("   ", "a post needs text")],
)
async def test_draft_post_reports_a_bad_post_as_a_result_not_an_error(text, reason):
    server = create_server(config_for("alice"), client=fake_chirp([]))
    async with Client(server) as client:
        result = await client.call_tool("draft_post", {"text": text}, raise_on_error=False)
    assert result.is_error is False
    assert result.data["valid"] is False
    assert result.data["reason"] == reason


async def test_exactly_the_limit_is_allowed():
    server = create_server(config_for("alice"), client=fake_chirp([]))
    async with Client(server) as client:
        data = (await client.call_tool("draft_post", {"text": "x" * MAX_POST_CHARS})).data
    assert data["valid"] is True and data["remaining"] == 0


async def test_publishing_is_refused_until_the_environment_enables_it():
    seen: list[httpx.Request] = []
    server = create_server(config_for("alice"), client=fake_chirp(seen))
    async with Client(server) as client:
        result = await client.call_tool("publish_post", {"text": "hi"}, raise_on_error=False)
    assert result.is_error is True
    assert "CHIRP_ALLOW_PUBLISH" in result.content[0].text
    # The refusal happens before any credential leaves the process.
    assert seen == []


async def test_publish_writes_with_the_token_the_session_returned():
    seen: list[httpx.Request] = []
    server = create_server(config_for("alice", CHIRP_ALLOW_PUBLISH="1"), client=fake_chirp(seen))
    async with Client(server) as client:
        data = (await client.call_tool("publish_post", {"text": "hello world"})).data
    assert seen[-1].headers["Authorization"] == f"Bearer chirp-token.{IDS['alice']}"
    assert data["account"] == "alice"
    assert data["user_id"] == IDS["alice"]


async def test_two_instances_of_the_same_code_write_to_different_accounts():
    for username in ("alice", "bob"):
        server = create_server(config_for(username, CHIRP_ALLOW_PUBLISH="1"), client=fake_chirp([]))
        async with Client(server) as client:
            data = (await client.call_tool("publish_post", {"text": f"hi from {username}"})).data
        assert data["user_id"] == IDS[username]


async def test_a_session_for_the_wrong_account_aborts_the_write():
    seen: list[httpx.Request] = []
    server = create_server(
        config_for("alice", CHIRP_ALLOW_PUBLISH="1"), client=fake_chirp(seen, session_username="bob")
    )
    async with Client(server) as client:
        result = await client.call_tool("publish_post", {"text": "hi"}, raise_on_error=False)
    assert result.is_error is True
    assert "bob" in result.content[0].text
    # The post call never happened.
    assert [r.url.path for r in seen] == ["/session"]


async def test_publish_rejects_text_the_service_would_reject():
    seen: list[httpx.Request] = []
    server = create_server(config_for("alice", CHIRP_ALLOW_PUBLISH="1"), client=fake_chirp(seen))
    async with Client(server) as client:
        result = await client.call_tool("publish_post", {"text": "x" * 281}, raise_on_error=False)
    assert result.is_error is True
    assert seen == []

"""Chirp's own guarantees, checked against a real instance on a temp data file."""

import httpx
import pytest

import chirpd


@pytest.fixture()
def chirp(tmp_path, monkeypatch):
    """A running Chirp with its own data file, torn down after each test."""
    monkeypatch.setattr(chirpd, "DATA_FILE", tmp_path / "chirp-data.json")
    server, url = chirpd.start_in_thread()
    yield url
    server.shutdown()
    server.server_close()


def signup(url: str, username: str) -> str:
    return httpx.post(f"{url}/accounts", json={"username": username}, timeout=5).json()["app_password"]


def token_for(url: str, username: str, password: str) -> str:
    return httpx.post(f"{url}/session", json={"username": username, "app_password": password}, timeout=5).json()["token"]


def test_an_app_password_is_four_groups_of_four(chirp):
    password = signup(chirp, "alice")
    assert [len(part) for part in password.split("-")] == [4, 4, 4, 4]


def test_two_accounts_get_different_ids_and_passwords(chirp):
    alice, bob = signup(chirp, "alice"), signup(chirp, "bob")
    assert alice != bob
    ids = {httpx.get(f"{chirp}/users/{u}", timeout=5).json()["user_id"] for u in ("alice", "bob")}
    assert len(ids) == 2


def test_a_username_cannot_be_taken_twice(chirp):
    signup(chirp, "alice")
    assert httpx.post(f"{chirp}/accounts", json={"username": "alice"}, timeout=5).status_code == 409


def test_the_wrong_app_password_is_rejected(chirp):
    signup(chirp, "alice")
    response = httpx.post(f"{chirp}/session", json={"username": "alice", "app_password": "nope"}, timeout=5)
    assert response.status_code == 401
    assert response.json()["error"] == "bad_credentials"


def test_a_post_is_attributed_to_the_token_holder(chirp):
    password = signup(chirp, "alice")
    signup(chirp, "bob")
    token = token_for(chirp, "alice", password)
    response = httpx.post(f"{chirp}/posts", json={"text": "mine"}, headers={"Authorization": f"Bearer {token}"}, timeout=5)
    assert response.json()["author"] == "alice"
    # Bob's timeline is untouched, which is the guarantee the sidecar relies on.
    assert httpx.get(f"{chirp}/users/bob", timeout=5).json()["post_count"] == 0


def test_a_post_without_a_token_is_refused(chirp):
    signup(chirp, "alice")
    assert httpx.post(f"{chirp}/posts", json={"text": "hi"}, timeout=5).status_code == 401


def test_the_service_enforces_its_own_length_limit(chirp):
    password = signup(chirp, "alice")
    token = token_for(chirp, "alice", password)
    response = httpx.post(f"{chirp}/posts", json={"text": "x" * 281}, headers={"Authorization": f"Bearer {token}"}, timeout=5)
    assert response.status_code == 400
    assert response.json()["error"] == "too_long"


def test_an_unknown_user_is_a_404(chirp):
    assert httpx.get(f"{chirp}/users/nobody", timeout=5).status_code == 404