mcp-pos-print/server.py

"""MCP server exposing the Epson TM-m30 receipt printer as a `print` tool.

Takes markdown, translates it to ESC/POS, sends raw bytes over TCP to the
printer at $POS_PRINTER_IP:$POS_PRINTER_PORT (default 192.168.0.184:9100).

See DECISIONS.md for design rationale and ~/bin/POS_PRINT.md for the printer
reference (Font B columns, native QR, troubleshooting).
"""

import asyncio
import os
import re
import socket
from datetime import datetime
from typing import Any
from zoneinfo import ZoneInfo

from escpos.printer import Dummy
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import TextContent, Tool

PRINTER_IP = os.environ.get("POS_PRINTER_IP", "192.168.0.184")
PRINTER_PORT = int(os.environ.get("POS_PRINTER_PORT", "9100"))
PRINTER_TIMEOUT = int(os.environ.get("POS_PRINTER_TIMEOUT", "10"))
COLS = 57  # Font B on 80mm paper. See POS_PRINT.md.
EASTERN = ZoneInfo("America/New_York")

server: Server = Server("pos-print")

TOOL_SCHEMA: dict[str, Any] = {
    "type": "object",
    "properties": {
        "markdown": {
            "type": "string",
            "description": (
                "The body to print, as markdown. Supports: # / ## / ### headings, "
                "**bold** and `code` inline, fenced ``` code blocks (no wrap), "
                "--- or *** horizontal rules, - / * / 1. lists, > blockquotes, "
                "[text](url) links, blank lines. Long lines word-wrap at 57 columns."
            ),
        },
        "title": {
            "type": "string",
            "description": "Optional bold centered header printed above the body.",
        },
        "timestamp": {
            "type": "boolean",
            "default": True,
            "description": "Print a date/time line under the title (Eastern Time).",
        },
        "qr_urls": {
            "type": "array",
            "items": {"type": "string"},
            "description": "Optional list of URLs to render as native hardware QR codes at the end.",
        },
        "cut": {
            "type": "boolean",
            "default": True,
            "description": "Auto-cut the paper at the end. Almost always leave true.",
        },
    },
    "required": ["markdown"],
}


@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="print",
            description=(
                "Print arbitrary markdown to the Epson TM-m30 receipt printer. "
                "Use for code snippets, summaries, TODO lists, QR codes, or any "
                "output the user wants on paper. Returns confirmation with byte "
                "count, or a clear error if the printer is unreachable."
            ),
            inputSchema=TOOL_SCHEMA,
        )
    ]


# ── Inline markdown parser ───────────────────────────────────────────────────
# Splits a line into [(style, text), ...] where style is one of:
#   "plain", "bold", "code"
# Supports: **bold**, `code`, [text](url) -> "text (url)"

_INLINE_RE = re.compile(
    r"(\*\*[^*\n]+?\*\*|`[^`\n]+?`|\[[^\]\n]+?\]\([^)\n]+?\))"
)


def parse_inline(line: str) -> list[tuple[str, str]]:
    """Parse a single line into styled segments. Order-preserving."""
    if not line:
        return [("plain", "")]
    out: list[tuple[str, str]] = []
    for part in _INLINE_RE.split(line):
        if not part:
            continue
        if part.startswith("**") and part.endswith("**"):
            out.append(("bold", part[2:-2]))
        elif part.startswith("`") and part.endswith("`"):
            out.append(("code", part[1:-1]))
        elif part.startswith("[") and "](" in part and part.endswith(")"):
            close = part.index("]")
            text = part[1:close]
            url = part[close + 2 : -1]
            out.append(("plain", f"{text} ({url})"))
        else:
            out.append(("plain", part))
    return out


# ── Word wrap that respects styled segments ──────────────────────────────────
def wrap_segments(
    segments: list[tuple[str, str]], width: int, indent: str = ""
) -> list[list[tuple[str, str]]]:
    """Wrap a styled line into multiple styled lines, never breaking mid-word
    where avoidable. First line uses no extra indent; continuation lines use
    `indent`. Returns a list of styled-segment lines."""
    lines: list[list[tuple[str, str]]] = [[]]
    col = 0
    first = True

    def line_width() -> int:
        return width if first else max(1, width - len(indent))

    for style, text in segments:
        # Split on whitespace but preserve the whitespace runs so we can rejoin.
        tokens = re.split(r"(\s+)", text)
        for tok in tokens:
            if not tok:
                continue
            if tok.isspace():
                # Whitespace: only add if not at start of line.
                if col > 0:
                    lines[-1].append((style, " "))
                    col += 1
                continue
            # Hard-break tokens longer than the line width by char.
            while len(tok) > line_width() - col and col > 0:
                lines.append([])
                col = 0
                first = False
            while len(tok) > line_width():
                head, tok = tok[: line_width()], tok[line_width() :]
                lines[-1].append((style, head))
                lines.append([])
                col = 0
                first = False
            lines[-1].append((style, tok))
            col += len(tok)

    # Apply continuation indent
    if indent and len(lines) > 1:
        for i in range(1, len(lines)):
            lines[i] = [("plain", indent)] + lines[i]

    # Drop trailing empty lines
    while lines and not lines[-1]:
        lines.pop()
    return lines or [[]]


def emit_styled_line(p: Dummy, segs: list[tuple[str, str]]) -> None:
    """Emit a single styled line (no trailing newline)."""
    for style, text in segs:
        if not text:
            continue
        if style == "bold":
            p.set(font="b", bold=True)
            p.text(text)
            p.set(font="b", bold=False)
        else:
            # 'code' and 'plain' both print as plain Font B; ESC/POS has no
            # monospace toggle (already monospace) and no background shading
            # worth using on a 57-col receipt.
            p.set(font="b", bold=False)
            p.text(text)


# ── Block-level markdown parser ──────────────────────────────────────────────
_HEADING_RE = re.compile(r"^(#{1,6})\s+(.+?)\s*$")
_HR_RE = re.compile(r"^\s*(?:-{3,}|\*{3,}|_{3,})\s*$")
_BULLET_RE = re.compile(r"^\s*[-*]\s+(.+?)\s*$")
_NUMBERED_RE = re.compile(r"^\s*(\d+)\.\s+(.+?)\s*$")
_QUOTE_RE = re.compile(r"^\s*>\s?(.*)$")
_FENCE_RE = re.compile(r"^\s*```")


def render_markdown(p: Dummy, md: str) -> None:
    """Walk markdown line by line and emit ESC/POS via the Dummy builder."""
    lines = md.splitlines()
    i = 0
    while i < len(lines):
        line = lines[i]

        # Fenced code block: print verbatim, no wrap, until closing fence.
        if _FENCE_RE.match(line):
            i += 1
            while i < len(lines) and not _FENCE_RE.match(lines[i]):
                p.set(font="b", bold=False, align="left")
                # Truncate over-width code lines with a continuation marker.
                text = lines[i]
                while len(text) > COLS:
                    p.text(text[: COLS - 1] + "»\n")  # » continuation marker
                    text = text[COLS - 1 :]
                p.text(text + "\n")
                i += 1
            i += 1  # consume the closing fence (or end of input)
            continue

        # Horizontal rule
        if _HR_RE.match(line):
            p.set(font="b", bold=False, align="left")
            p.text("-" * COLS + "\n")
            i += 1
            continue

        # Headings
        m = _HEADING_RE.match(line)
        if m:
            level = len(m.group(1))
            text = m.group(2)
            if level == 1:
                p.set(font="b", bold=True, align="center", width=2, height=2)
                p.text(text + "\n")
                p.set(font="b", bold=False, align="left", width=1, height=1)
            elif level == 2:
                p.set(font="b", bold=True, align="left")
                p.text(text + "\n")
                p.set(font="b", bold=False, align="left")
            else:
                p.set(font="b", bold=True, align="left")
                p.text(text + "\n")
                p.set(font="b", bold=False, align="left")
            i += 1
            continue

        # Blockquote
        m = _QUOTE_RE.match(line)
        if m:
            inner = m.group(1)
            segs = parse_inline(inner)
            wrapped = wrap_segments(segs, COLS - 2, indent="  ")
            for wline in wrapped:
                p.set(font="b", bold=False, align="left")
                p.text("| ")
                emit_styled_line(p, wline)
                p.text("\n")
            i += 1
            continue

        # Bullet list item
        m = _BULLET_RE.match(line)
        if m:
            inner = m.group(1)
            segs = parse_inline(inner)
            wrapped = wrap_segments(segs, COLS - 2, indent="  ")
            for j, wline in enumerate(wrapped):
                p.set(font="b", bold=False, align="left")
                p.text("* " if j == 0 else "")
                emit_styled_line(p, wline)
                p.text("\n")
            i += 1
            continue

        # Numbered list item
        m = _NUMBERED_RE.match(line)
        if m:
            num = m.group(1)
            inner = m.group(2)
            prefix = f"{num}. "
            segs = parse_inline(inner)
            wrapped = wrap_segments(segs, COLS - len(prefix), indent=" " * len(prefix))
            for j, wline in enumerate(wrapped):
                p.set(font="b", bold=False, align="left")
                p.text(prefix if j == 0 else "")
                emit_styled_line(p, wline)
                p.text("\n")
            i += 1
            continue

        # Blank line
        if not line.strip():
            p.set(font="b", bold=False, align="left")
            p.text("\n")
            i += 1
            continue

        # Plain paragraph line — wrap and emit with inline styles.
        segs = parse_inline(line)
        wrapped = wrap_segments(segs, COLS)
        for wline in wrapped:
            p.set(font="b", bold=False, align="left")
            emit_styled_line(p, wline)
            p.text("\n")
        i += 1


# ── Receipt builder ──────────────────────────────────────────────────────────
def build_receipt(
    markdown: str,
    title: str | None,
    timestamp: bool,
    qr_urls: list[str] | None,
    cut: bool,
) -> bytes:
    p = Dummy(profile="default")

    if title:
        p.set(font="b", bold=True, align="center", width=2, height=2)
        p.text(title + "\n")
        p.set(font="b", bold=False, align="center", width=1, height=1)
    if timestamp:
        p.set(font="b", bold=False, align="center")
        p.text(datetime.now(EASTERN).strftime("%a, %b %d  %H:%M ET") + "\n")
    if title or timestamp:
        p.set(font="b", bold=False, align="left")
        p.text("=" * COLS + "\n")

    render_markdown(p, markdown)

    if qr_urls:
        p.set(font="b", bold=False, align="left")
        p.text("-" * COLS + "\n")
        for url in qr_urls:
            p.set(font="b", bold=False, align="center")
            p.qr(url, native=True, size=6)
            p.set(font="b", bold=False, align="center")
            p.text(url + "\n")
            p.text("\n")

    if cut:
        p.set(font="b", bold=False, align="left")
        p.text("\n")
        p.cut()

    return p.output


def send_bytes(raw: bytes) -> None:
    """Send raw ESC/POS bytes to the printer over TCP. Raises on failure."""
    with socket.create_connection(
        (PRINTER_IP, PRINTER_PORT), timeout=PRINTER_TIMEOUT
    ) as sock:
        sock.sendall(raw)


@server.call_tool()
async def call_tool(name: str, args: dict[str, Any]) -> list[TextContent]:
    if name != "print":
        return [TextContent(type="text", text=f"Error: unknown tool `{name}`.")]

    markdown = args.get("markdown")
    if not markdown or not str(markdown).strip():
        return [TextContent(type="text", text="Error: `markdown` is empty — nothing to print.")]

    title = args.get("title")
    timestamp = bool(args.get("timestamp", True))
    qr_urls = args.get("qr_urls") or []
    cut = bool(args.get("cut", True))

    def _build_and_send() -> tuple[int | None, str | None]:
        try:
            raw = build_receipt(str(markdown), title, timestamp, list(qr_urls), cut)
        except Exception as e:
            return None, f"Error: failed to build receipt bytes: {type(e).__name__}: {e}"
        try:
            send_bytes(raw)
        except (TimeoutError, socket.timeout):
            return (
                None,
                f"Error: printer at {PRINTER_IP}:{PRINTER_PORT} timed out after "
                f"{PRINTER_TIMEOUT}s. Likely off, paper out, or ARP issue — ping first.",
            )
        except OSError as e:
            return (
                None,
                f"Error: cannot reach printer at {PRINTER_IP}:{PRINTER_PORT} "
                f"({e}). Check power, network, paper.",
            )
        return len(raw), None

    byte_count, err = await asyncio.to_thread(_build_and_send)
    if err:
        return [TextContent(type="text", text=err)]
    return [
        TextContent(
            type="text",
            text=f"Printed {byte_count} bytes to {PRINTER_IP}:{PRINTER_PORT}.",
        )
    ]


async def main() -> None:
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())


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