Mortdecai
421c3c9660
Single tool `print(markdown, title?, timestamp?, qr_urls?, cut?)`. Hand-rolled line-by-line markdown parser translates headings/bold/code/lists/blockquotes/ HR/links to ESC/POS, then sends raw bytes via TCP to 192.168.0.184:9100. Mirrors mcp-gemma4 layout (stdio server, single server.py, mcp package).
404 lines
14 KiB
Python
404 lines
14 KiB
Python
"""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())
|