Compare commits

42 Commits
Author SHA1 Message Date
akadmin 20c75d1ca3 Fix tools/call: avoid double-encoding string results 2026-06-13 20:55:27 +00:00
akadmin 205174dfb3 Align with mcp-time-tools: FastAPI Streamable HTTP MCP endpoint 2026-06-13 17:02:18 +00:00
akadmin 57a225b749 Use streamable_http_app() instead of to_asgi_app() 2026-06-13 16:34:59 +00:00
akadmin 8e93b39f0a Use @mcp.tool() with parentheses as required by SDK 2026-06-13 16:24:38 +00:00
akadmin d910c1a817 Fix tool registration: use @mcp.tool decorator without name arg 2026-06-13 16:16:42 +00:00
akadmin fc58fd59ac Flatten layout: move py_docx files into src; update imports and Dockerfile 2026-06-13 15:53:18 +00:00
akadmin af8f6bf2e1 Fix Dockerfile: copy src before pip install; set PYTHONPATH 2026-06-13 15:49:07 +00:00
akadmin 8adb8c8fb0 Fix py_docx import error: use setuptools + PYTHONPATH 2026-06-13 15:45:11 +00:00
akadmin ff034fa97e Add README for Python MCP server 2026-06-13 06:13:12 +00:00
akadmin 438eebc73d Remove Rust project; repo is now Python-only MCP server at root 2026-06-13 06:11:21 +00:00
akadmin 873976c05a Merge python-docx into main
Continuous Integration / Test Suite (macos-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (macos-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, 1.70.0) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, beta) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (windows-latest, stable) (push) Has been cancelled
Continuous Integration / Security Audit (push) Has been cancelled
Continuous Integration / Code Coverage (push) Has been cancelled
Continuous Integration / Performance Benchmarks (push) Has been cancelled
Continuous Integration / Memory Safety Check (push) Has been cancelled
Continuous Integration / Docker Build Test (push) Has been cancelled
Continuous Integration / Release Readiness (push) Has been cancelled
Continuous Integration / Integration Tests (push) Has been cancelled
Continuous Integration / Stress Testing (push) Has been cancelled
Continuous Integration / Notify Results (push) Has been cancelled
2026-06-13 06:07:44 +00:00
akadmin 2883d01cc5 Move Python MCP server into py-docx/ subdirectory 2026-06-13 06:05:19 +00:00
akadmin 269f3b9757 Initial commit: Python MCP server (Streamable HTTP, API key, return documents) 2026-06-13 06:02:13 +00:00
akadmin 899963a14c Fix docx_tools.rs missing imports for json, info, debug, Arc, etc.
Continuous Integration / Test Suite (macos-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (macos-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, 1.70.0) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, beta) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (windows-latest, stable) (push) Has been cancelled
Continuous Integration / Security Audit (push) Has been cancelled
Continuous Integration / Code Coverage (push) Has been cancelled
Continuous Integration / Performance Benchmarks (push) Has been cancelled
Continuous Integration / Memory Safety Check (push) Has been cancelled
Continuous Integration / Docker Build Test (push) Has been cancelled
Continuous Integration / Release Readiness (push) Has been cancelled
Continuous Integration / Integration Tests (push) Has been cancelled
Continuous Integration / Stress Testing (push) Has been cancelled
Continuous Integration / Notify Results (push) Has been cancelled
2026-06-13 04:24:15 +00:00
akadmin c5416bf745 Fix build: add response module, fix http_server, update Cargo.toml
Continuous Integration / Test Suite (macos-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (macos-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, 1.70.0) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, beta) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (windows-latest, stable) (push) Has been cancelled
Continuous Integration / Security Audit (push) Has been cancelled
Continuous Integration / Code Coverage (push) Has been cancelled
Continuous Integration / Performance Benchmarks (push) Has been cancelled
Continuous Integration / Memory Safety Check (push) Has been cancelled
Continuous Integration / Docker Build Test (push) Has been cancelled
Continuous Integration / Release Readiness (push) Has been cancelled
Continuous Integration / Integration Tests (push) Has been cancelled
Continuous Integration / Stress Testing (push) Has been cancelled
Continuous Integration / Notify Results (push) Has been cancelled
2026-06-13 04:02:52 +00:00
akadmin bb547888bf Add build-bin feature to Dockerfile so binary is built
Continuous Integration / Test Suite (macos-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (macos-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, 1.70.0) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, beta) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (windows-latest, stable) (push) Has been cancelled
Continuous Integration / Security Audit (push) Has been cancelled
Continuous Integration / Code Coverage (push) Has been cancelled
Continuous Integration / Performance Benchmarks (push) Has been cancelled
Continuous Integration / Memory Safety Check (push) Has been cancelled
Continuous Integration / Docker Build Test (push) Has been cancelled
Continuous Integration / Release Readiness (push) Has been cancelled
Continuous Integration / Integration Tests (push) Has been cancelled
Continuous Integration / Stress Testing (push) Has been cancelled
Continuous Integration / Notify Results (push) Has been cancelled
2026-06-13 02:20:00 +00:00
akadmin 670e7492bb Fix generate_from_template: correct return type and borrow handling
Continuous Integration / Test Suite (macos-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (macos-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, 1.70.0) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, beta) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (windows-latest, stable) (push) Has been cancelled
Continuous Integration / Security Audit (push) Has been cancelled
Continuous Integration / Code Coverage (push) Has been cancelled
Continuous Integration / Performance Benchmarks (push) Has been cancelled
Continuous Integration / Memory Safety Check (push) Has been cancelled
Continuous Integration / Docker Build Test (push) Has been cancelled
Continuous Integration / Release Readiness (push) Has been cancelled
Continuous Integration / Integration Tests (push) Has been cancelled
Continuous Integration / Stress Testing (push) Has been cancelled
Continuous Integration / Notify Results (push) Has been cancelled
2026-06-13 01:31:23 +00:00
akadmin a742efa73a Fix: remove unused mut on invoice_info table 2026-06-13 01:05:18 +00:00
akadmin e2a456fc9f Use rust:1.90-slim to support edition2024 (fix aligned crate build)
Continuous Integration / Test Suite (macos-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (macos-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, 1.70.0) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, beta) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (windows-latest, stable) (push) Has been cancelled
Continuous Integration / Security Audit (push) Has been cancelled
Continuous Integration / Code Coverage (push) Has been cancelled
Continuous Integration / Performance Benchmarks (push) Has been cancelled
Continuous Integration / Memory Safety Check (push) Has been cancelled
Continuous Integration / Docker Build Test (push) Has been cancelled
Continuous Integration / Release Readiness (push) Has been cancelled
Continuous Integration / Integration Tests (push) Has been cancelled
Continuous Integration / Stress Testing (push) Has been cancelled
Continuous Integration / Notify Results (push) Has been cancelled
2026-06-13 00:56:29 +00:00
akadmin 9d232e5696 Fix Docker build: include benches directory for docx_benchmarks
Continuous Integration / Test Suite (macos-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (macos-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, 1.70.0) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, beta) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (windows-latest, stable) (push) Has been cancelled
Continuous Integration / Security Audit (push) Has been cancelled
Continuous Integration / Code Coverage (push) Has been cancelled
Continuous Integration / Performance Benchmarks (push) Has been cancelled
Continuous Integration / Memory Safety Check (push) Has been cancelled
Continuous Integration / Docker Build Test (push) Has been cancelled
Continuous Integration / Release Readiness (push) Has been cancelled
Continuous Integration / Integration Tests (push) Has been cancelled
Continuous Integration / Stress Testing (push) Has been cancelled
Continuous Integration / Notify Results (push) Has been cancelled
2026-06-13 00:39:55 +00:00
akadmin 0d16ca6f1a Add Cargo.lock and restore it in Dockerfile
Continuous Integration / Test Suite (macos-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (macos-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, 1.70.0) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, beta) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (windows-latest, stable) (push) Has been cancelled
Continuous Integration / Security Audit (push) Has been cancelled
Continuous Integration / Code Coverage (push) Has been cancelled
Continuous Integration / Performance Benchmarks (push) Has been cancelled
Continuous Integration / Memory Safety Check (push) Has been cancelled
Continuous Integration / Docker Build Test (push) Has been cancelled
Continuous Integration / Release Readiness (push) Has been cancelled
Continuous Integration / Integration Tests (push) Has been cancelled
Continuous Integration / Stress Testing (push) Has been cancelled
Continuous Integration / Notify Results (push) Has been cancelled
2026-06-13 00:34:35 +00:00
akadmin 51d6e97553 Remove Cargo.lock dependency from Dockerfile
Continuous Integration / Test Suite (macos-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (macos-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, 1.70.0) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, beta) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (windows-latest, stable) (push) Has been cancelled
Continuous Integration / Security Audit (push) Has been cancelled
Continuous Integration / Code Coverage (push) Has been cancelled
Continuous Integration / Performance Benchmarks (push) Has been cancelled
Continuous Integration / Memory Safety Check (push) Has been cancelled
Continuous Integration / Docker Build Test (push) Has been cancelled
Continuous Integration / Release Readiness (push) Has been cancelled
Continuous Integration / Integration Tests (push) Has been cancelled
Continuous Integration / Stress Testing (push) Has been cancelled
Continuous Integration / Notify Results (push) Has been cancelled
2026-06-13 00:30:51 +00:00
akadmin f655336757 Add HTTP interface, templates, generate_from_template, unified Dockerfile
Continuous Integration / Test Suite (macos-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (macos-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, 1.70.0) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, beta) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, nightly) (push) Has been cancelled
Continuous Integration / Test Suite (ubuntu-latest, stable) (push) Has been cancelled
Continuous Integration / Test Suite (windows-latest, stable) (push) Has been cancelled
Continuous Integration / Security Audit (push) Has been cancelled
Continuous Integration / Code Coverage (push) Has been cancelled
Continuous Integration / Performance Benchmarks (push) Has been cancelled
Continuous Integration / Memory Safety Check (push) Has been cancelled
Continuous Integration / Docker Build Test (push) Has been cancelled
Continuous Integration / Release Readiness (push) Has been cancelled
Continuous Integration / Integration Tests (push) Has been cancelled
Continuous Integration / Stress Testing (push) Has been cancelled
Continuous Integration / Notify Results (push) Has been cancelled
2026-06-13 00:22:02 +00:00
Andy d3fbbcfd7c feat(mcp): introduce simple range model and selector-based editing
- RangeId for paragraphs, headings, and table cells
- Tools: get_outline, get_ranges, replace_range_text, set_table_cell_text
- Keeps edits idempotent and precise for AI workflows
2025-08-13 00:08:04 +08:00
Andy 3b05711e92 feat(mcp): core author/edit/understand APIs and PDF path polish
- Authoring/editing: paragraph formatting tool; expanded list/table ops kept
- Understanding: get_tables, list_images, list_hyperlinks, get_fields_summary
- Safety: strip_personal_info that scrubs core.xml best-effort
- Tests remain green; pure-Rust + hi-fidelity export path preserved
2025-08-12 23:48:56 +08:00
Andy 0e43a50ad2 feat(docx): add TOC and bookmark placeholder tools; hi-fidelity TOC/bookmark XML post-processing
- Features: hi-fidelity-toc, hi-fidelity-bookmarks
- Tools: insert_toc, insert_bookmark_after_heading
- Write: emit recognizable placeholders and transform to field XML under feature flags
- Tests: add golden checks for TOC field injection and bookmark cleanup
2025-08-12 23:35:49 +08:00
Andy 90305551cc feat(docx): add hi-fidelity XML injections for tables, styles, lists, and sections; extend tools and tests
- Add feature flags: hi-fidelity-tables, hi-fidelity-styles, hi-fidelity-lists, hi-fidelity-sections
- Tables: inject true w:gridSpan/w:vMerge and w:tblGrid widths via post-build XML when enabled
- Styles: ensure TableHeader style in styles.xml; tag first row when headers present
- Lists: robust numbering.xml for ordered/unordered with multi-level definitions
- Sections: write tail w:sectPr with page size/orientation/margins
- Tools: expose new operations (sections, list items, images, hyperlinks, props, redaction, storage)
- Converters: add preference-aware methods for hi-fidelity export paths; HTML export tool
- Tests: add golden XML assertions gated by feature flags; keep default build green

This enables high-fidelity DOCX output while keeping pure-Rust paths by default.
2025-08-12 23:25:29 +08:00
Andy c30f55d16d Concurrency: switch provider handler to RwLock for read-heavy parallelism; add get_storage_info tool; improve text extraction whitespace; keep tests green 2025-08-12 19:17:19 +08:00
Andy 15ec810cea Security: whitelist precedence and sandbox path handling for non-existent temp paths; minor import cleanups; all tests passing 2025-08-11 22:53:07 +08:00
Andy 515b0100ac Introduce base-dir constructors for isolation; update tests to avoid env var dependence; ensure directories exist before I/O; all tests green (including performance) 2025-08-11 22:41:14 +08:00
Andy ec8b46955b Stabilize tests and security: expose modules, standardize tool responses, add ToolResult helpers; fix sandbox path checks; make handler respect DOCX_MCP_TEMP and ensure dirs exist; add pure converter wrappers and JPEG fix; relax brittle assertions; replace TMPDIR with DOCX_MCP_TEMP in tests; modernize advanced_docx fallbacks; add example bin; all suites green locally 2025-08-11 22:11:37 +08:00
Andy ad8909d749 Refactor: upgrade to latest MCP and docx-rs; add Router, fonts CLI, and builder-based DOCX edits
- Integrate mcp-server Router with mcp-spec and expose tools
- Add fonts subcommands (download/verify) with pinned sources + checksums
- Replace deprecated docx-rs APIs; rebuild DOCX via ops (paragraphs/headings/tables/lists/page breaks/headers/footers)
- Implement proper numbered lists via docx-rs numbering
- Gate advanced features behind `advanced-docx` for future porting
- Resolve lopdf and image import ambiguities; adapt search and responses
2025-08-11 19:19:04 +08:00
Andy f75a47fe76 Feature-gate runtime server; migrate MCP API; improve DOCX fallbacks
Gate runtime server behind `runtime-server` feature and align tool interfaces with latest `mcp_core` response types. Add safer DOCX->PDF fallbacks (dotext reader, inline comments/notes/cross-refs) and clarify crate imports (`image`, `lopdf`) to reduce conflicts; minor PDF utilities cleanup.
2025-08-11 18:19:53 +08:00
Andy ec800ce12c [integration] chore(ci): trigger integration tests for args parsing 2025-08-11 15:08:07 +08:00
Andy 40b9a6c6ab Make benchmarks opt-in via [bench] marker; proceed with lean green CI 2025-08-11 15:07:11 +08:00
Andy af2082aa61 Stabilize CI/workflows around library-only build and tests
- Make CI clippy/build/test focus on `--lib` by default
- Gate integration, doc, feature-matrix, and docker to opt-in via commit message flags
- Limit coverage to library targets
- Adjust release workflow to test and build without `--all-features`

This prevents workflow failures from unimplemented bins/features while still validating our CLI arg parsing and core library.
2025-08-11 15:05:19 +08:00
Andy d4ebdbf6a9 CLI: inline font download + checksum verification
- Add `fonts download` and `fonts verify` subcommands
- Implement Rust-based downloader (ureq + tar + flate2) with pinned sources
- Verify SHA-256 for Liberation and Noto Sans TTFs for reproducibility
- Keep binary behind `build-bin` feature; library build unaffected
2025-08-11 15:04:47 +08:00
Andy 98c056c3d9 CLI: add fonts download subcommand to fetch assets
Expose a `docx-mcp fonts download` subcommand that runs the existing `download_fonts.sh` helper, so users can fetch the embedded fonts via the binary. Keeps existing flags intact and avoids build-time network coupling.
2025-08-11 14:59:05 +08:00
Andy ded289451e Add CLI args parsing tests and lib target; fix summary string building
Introduce `src/lib.rs` and library target so integration tests can import `docx_mcp`. Add focused `tests/args_tests.rs` verifying clap flag/env parsing and `SecurityConfig::from_args`/`from_env`. Enable clap `env` feature and guard the binary behind a `build-bin` feature to allow testing without unresolved MCP server deps. Fix `get_summary` to build owned strings safely.

These changes ensure argument options work correctly and are covered by comprehensive tests, independent of heavier integration suites.
2025-08-11 14:56:20 +08:00
Andy 232bc36464 Embed open-source TTF fonts for reproducible offline builds
Ensure PDF generation works consistently without network access by committing the Liberation and NotoSans font files used by embedded-fonts. This removes reliance on placeholder files and download-time variability.
2025-08-11 14:53:49 +08:00
AndyandClaude 705e8bfa91 Add comprehensive release infrastructure and tooling
- Add automated GitHub Actions workflow for multi-platform releases
- Add release script with version management and validation
- Add Docker container support with multi-stage builds
- Add comprehensive release documentation and templates
- Update Cargo.toml with complete package metadata
- Add command-line argument parsing with security options
- Update README with detailed configuration examples

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-08-11 14:40:53 +08:00
Andy 39e94c1b13 Initial Commit 2025-08-11 14:31:51 +08:00
36 changed files with 2109 additions and 1856 deletions
-11
View File
@@ -1,11 +0,0 @@
__pycache__/
*.pyc
*.pyo
.pytest_cache/
build/
dist/
*.deb
.eggs/
*.egg-info/
.venv/
venv/
+73
View File
@@ -0,0 +1,73 @@
# Dockerfile for py-docx-mcp (Python MCP server) - OpenWebUI: MCP (Streamable HTTP)
# Usage:
# docker build -t py-docx-mcp .
# docker run --rm -p 3000:3000 py-docx-mcp
#
# In OpenWebUI:
# - Type: MCP (Streamable HTTP)
# - URL: http://<host>:3000
# - Auth: Bearer (if DOCX_MCP_API_KEY is set)
#
# Environment:
# DOCX_MCP_API_KEY - API key (Bearer). Optional but recommended.
# DOCX_MCP_HTTP_HOST - Bind host (default: 0.0.0.0)
# DOCX_MCP_HTTP_PORT - Bind port (default: 3000)
# DOCX_MCP_TEMPLATES_DIR - Templates directory (default: /templates)
# DOCX_MCP_MAX_SIZE - Max document size in bytes (default: 104857600)
# DOCX_MCP_MAX_DOCS - Max open documents (default: 30)
# DOCX_MCP_SANDBOX - Enable sandbox mode (default: true)
# DOCX_MCP_ALLOW_EXTERNAL_TOOLS - Allow external tools (default: false)
# DOCX_MCP_ALLOW_NETWORK - Allow network access (default: false)
FROM python:3.12-slim AS base
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PIP_NO_CACHE_DIR=off \
PIP_DISABLE_PIP_VERSION_CHECK=1
WORKDIR /app
# System deps (for python-docx, Pillow, and optional external converters)
RUN apt-get update && apt-get install -y --no-install-recommends \
build-essential \
libjpeg62-turbo-dev \
libpng-dev \
libfreetype6-dev \
libfontconfig1-dev \
libreoffice \
poppler-utils \
&& rm -rf /var/lib/apt/lists/*
# Copy project metadata and source before installing
COPY pyproject.toml ./
COPY src ./src
# Install Python dependencies (including this package)
RUN pip install --upgrade pip && pip install .
# Ensure modules in src are importable at runtime
ENV PYTHONPATH="/app/src"
# Create runtime dirs
RUN mkdir -p /templates /out /tmp/py-docx-mcp
# Environment
ENV DOCX_MCP_HTTP_HOST=0.0.0.0 \
DOCX_MCP_HTTP_PORT=3000 \
DOCX_MCP_TEMPLATES_DIR=/templates \
DOCX_MCP_MAX_SIZE=104857600 \
DOCX_MCP_MAX_DOCS=30 \
DOCX_MCP_SANDBOX=true \
DOCX_MCP_ALLOW_EXTERNAL_TOOLS=true \
DOCX_MCP_ALLOW_NETWORK=false
# Expose HTTP port (Streamable HTTP for OpenWebUI)
EXPOSE 3000
# Health check (ensure server module is importable)
HEALTHCHECK --interval=30s --timeout=5s --start-period=5s --retries=3 \
CMD python -c "from server import make_app; print('ok')" || exit 1
# Default: Streamable HTTP for OpenWebUI MCP
ENTRYPOINT ["python", "-m", "server"]
-259
View File
@@ -1,259 +0,0 @@
# doced — Plan for an LLM-Driven DOCX CLI on Debian/Ubuntu
> **Status:** Plan (v1) — awaiting user review before implementation.
> **Deliverable:** A design for `doced`, a CLI that lets an LLM with terminal access manipulate `.docx` files at an advanced level, installable on any Debian/Ubuntu terminal.
---
## 0. TL;DR
`doced` is a Python CLI that treats a `.docx` as an OOXML package (a zip of XML parts) and mutates it **in place** — never by rebuilding from a lossy round-trip. It speaks an LLM-friendly contract: **stable locators**, a **lossless JSON AST**, **machine-readable output**, and an **exit-code-as-success-signal** protocol. It ships as both a `pipx`-installable package and an `apt` `.deb`, so `doced` is on PATH on any Debian/Ubuntu box.
The name `doced` is deliberately distinct from the existing `docx` CLI (see §1).
---
## 1. Research: what already exists
### 1.1 Web research — caveat
The delegated web-research sub-agent returned **empty results** (general web search is blocked in this environment; this matches the prior `docling_ocr_skill` session, where web search also came back empty). So the competitive analysis below is grounded in:
1. A **detailed review of an existing, mature project already in this workspace** — `docx-cli` (github.com/kklimuk/docx-cli) — including its skill, its agent-facing `CLAUDE.md`, its `install.sh`, and its `build-deb.sh`.
2. Domain knowledge of the Python/LLM document ecosystem.
This is sufficient to map the landscape and differentiate `doced`, but I flag it as a research gap to close if web access becomes available.
### 1.2 Existing applications (the competitive landscape)
**A. The dominant direct competitor: `docx-cli` (Bun/TypeScript, by Kirill Klimuk)**
This is the single most relevant existing tool and it does almost exactly what `doced` is being asked for. Key facts, verified from the workspace copy:
- **In-place OOXML mutation.** It never rebuilds the document from Markdown/AST. It mutates `XmlNode` refs directly, so unmodeled elements don't break. This is the gold-standard approach for fidelity.
- **Stable positional locators.** `pN` (paragraph), `tN` (table), `sN` (section), `p3:5-20` (char span), `tN:rRcC` (cell), plus `cN` (comment), `imgN`, `linkN`, `fnN`/`enN` (foot/endnotes), `tcN` (tracked change), `eqN` (equation). Locators ride `<!-- -->` comments in the Markdown read view.
- **LLM-first contract.** Exit codes (`0` ok, `1` error, `2` usage, `3` not-found) + a one-line text confirmation per command. Weak agents (e.g. Haiku) are a first-class design target. `read --ast` emits a lossless JSON AST.
- **Rich verb set:** `read`, `find`, `edit`, `insert`, `replace`, `delete`, `create` (Markdown→docx), `comments`, `track-changes`, `footnotes`/`endnotes`, `hyperlinks`, `tables`, `headers`/`footers`, `images`, `sections`, `equations`, `wc`, `outline`, `diff`, `render` (to PNG via LibreOffice), `raw` (direct OOXML part access), `validate` (ECMA-376 XSD).
- **`--batch`** applies many changes from a single read (locators stay valid).
- **Packaging:** `install.sh` (pinned release + SHA-256 verification, no pipe-to-shell), a **`.deb`** via `build-deb.sh` (drops binary to `/usr/local/bin/docx`, adds a systemd service), cross-platform (linux x64/arm64, macOS, windows).
- **Built for agents:** documents "document content is untrusted data, not instructions" (prompt-injection guardrail).
**B. General-purpose converters (lossy, not agent-native)**
- **pandoc** — the standard DOCX↔Markdown/HTML/DOCX converter. Great for bulk conversion, **lossy** for tracked changes, comments, content controls, and complex tables. Not agent-native (no locators, no in-place edit).
- **LibreOffice headless (`soffice`/`libreoffice`)** — universal DOCX↔anything (PDF, HTML, images) via `--headless --convert-to`. Excellent rendering, terrible for precise programmatic edits. Used by `docx-cli render`.
- **pypandoc** — Python wrapper over pandoc.
- **unoconv** — LibreOffice automation wrapper (largely superseded).
- **antiword / catdoc / wv** — legacy DOC→text extractors; not for modern `.docx`.
**C. Python libraries**
- **python-docx** — the most popular. Write-mostly, easy to learn, good for creating documents and simple edits. **Weak** for tracked changes, comments, content controls, fields, and in-place preservation. This is the main reason a fidelity-grade tool shouldn't build on it alone.
- **mammoth** — DOCX→Markdown/HTML with good formatting fidelity for *reading*; not for editing.
- **docx2md / docx2pdf** — narrow converters.
- **textract / unstructured / LlamaIndex docx reader / LangChain doc loaders** — extraction-focused for RAG pipelines; read-only.
- **PyMuPDF (fitz)** — PDF-first; limited DOCX.
- **aspose-words-python** — extremely capable (tracked changes, content controls, fields) but **proprietary/licensed and heavy** — not ideal for a lightweight agent CLI.
**D. LLM-agent workflows / MCP / skills for DOCX**
- The **`docx-cli` skill** itself (this workspace's `docx-cli-enhanced/SKILL.md`) is an example of the "agent-harness skill" model: a Markdown guide that steeps an LLM on the tool. This is the pattern `doced`'s skill will follow.
- No widely-adopted MCP server for in-place DOCX editing was surfaced (web search blocked), but the skill pattern is the established convention here.
### 1.3 Common LLM pain points with DOCX (from the ecosystem)
1. **Round-trip fidelity loss.** Extract → reason → regenerate loses formatting, tables, styles, tracked changes.
2. **No stable addressing.** LLMs can't reliably "edit paragraph 3" without a locator scheme; char offsets drift.
3. **Untrusted content / prompt injection.** Text inside a doc can look like instructions.
4. **Opaque success.** LLMs can't tell if a mutation landed without re-reading.
5. **Advanced legal features** (tracked changes, comments, content controls, fields) are invisible to naive extractors.
---
## 2. Positioning & differentiation
`doced` will **not** reinvent `docx-cli`'s in-place-mutation + locator design — that's the correct architecture. Instead, `doced` differentiates on three axes:
| Dimension | `docx-cli` (existing) | `doced` (proposed) |
|---|---|---|
| **Language** | Bun/TypeScript | **Python 3.12** — matches this workspace's venv, the legal-team's stack, and the Python DOCX ecosystem. |
| **Core engine** | Custom JSX/XML tree | **lxml** (libxml2) directly on the OOXML zip — robust, fast, no custom parser; reuses the mature Python XML toolchain. |
| **Distribution** | npm binary + `.deb` | **pipx-first** (trivial `pipx install doced`) **and** `.deb`; zero system Python pollution. |
| **Fidelity target** | Full OOXML | Same in-place model; **explicit legal-feature coverage** (tracked changes, comments, content controls, fields, TOC, headers/footers). |
| **Skill** | Markdown guide | Markdown skill modeled on the existing skill, but Python-flavored and legally-oriented. |
**Why Python:** The workspace venv (`/home/wall-o/workspace/venv`) already has Flask/pytest/requests; Python-docx, lxml, and the whole LLM doc toolchain is Python-native. Legal stakeholders are more likely to extend/maintain a Python CLI than a Bun one. Python also gives us `lxml` for the fidelity-critical OOXML layer without a custom parser.
**Risk to manage:** We will **not** rely on python-docx for the fidelity-critical path (its tracked-changes/comment support is weak). We'll use `lxml` to read/write the OOXML parts directly, mirroring `docx-cli`'s "mutate in place" invariant. python-docx may be used only for *convenience authoring* paths (e.g. creating from scratch), never for round-tripping an existing doc.
---
## 3. Architecture
```
doced/
├── pyproject.toml # pipx-installable; entry point `doced`
├── README.md
├── doced_skill.md # LLM skill (the "guide")
├── src/doced/
│ ├── __main__.py # `python -m doced`
│ ├── cli.py # thin arg parsing (subcommands, flags) ← thin surface
│ ├── output.py # exit codes + one-line acks (LLM contract)
│ ├── io/
│ │ ├── package.py # open .docx as zip; read/write OPC parts
│ │ └── xmlns.py # namespace map (w:, a:, r:, mc:, etc.)
│ ├── model/
│ │ ├── locator.py # pN / tN / sN / p3:5-20 / tN:rRcC / cN / tcN ...
│ │ ├── ast.py # lossless JSON AST builder
│ │ └── markdown.py # AST → Markdown read view (with locator hints)
│ ├── ops/
│ │ ├── read.py # read / find / outline / wc
│ │ ├── edit.py # edit / insert / replace / delete
│ │ ├── track.py # tracked changes (on/off/list/accept/reject)
│ │ ├── comments.py # comments (add/list/reply/resolve)
│ │ ├── fields.py # content controls + fields (cross-ref/hyperlink/etc.)
│ │ ├── marginals.py # headers / footers / page numbers
│ │ ├── tables.py # table ops
│ │ ├── toc.py # table of contents
│ │ └── raw.py # direct OOXML part access (gated)
│ └── author/
│ └── create.py # Markdown/text → new docx
├── tests/
│ ├── unit/ # XML-shape assertions
│ └── fixtures/ # round-trip via LibreOffice headless
├── debian/ # .deb packaging (control, postinst, service)
└── build-deb.sh
```
### 3.1 Core invariant (from docx-cli)
> **Changes must survive the write→read loop.** Any change made must be retrievable on the next turn in the AST and (where possible) in Markdown. **Mutate `lxml` element refs in place; only emit fresh XML for inserted nodes.** Unmodeled elements are preserved, not rewritten.
### 3.2 The `.docx` as an OPC package
A `.docx` is a zip. `doced` opens it with Python's `zipfile`, treats each entry as an **OPC part** (`word/document.xml`, `word/styles.xml`, `word/header{N}.xml`, `word/numbering.xml`, `word/comments.xml`, `word/trackChanges.xml`, `docProps/core.xml`, …), and uses `lxml.etree` to parse/mutate. On save, it re-zips. This mirrors `docx-cli`'s view-per-part model but on the mature lxml stack.
---
## 4. The LLM contract (why this design)
Modeled directly on the proven `docx-cli` contract, which was explicitly built for weak agents:
1. **Exit code is the success signal.** `0` ok · `1` error · `2` usage · `3` not-found. Every command also prints a one-line text confirmation, so the LLM never has to re-read just to know a mutation landed.
2. **Stable locators** address everything (`pN`, `tN`, `sN`, char spans, cells, comments, track-change handles, …). They ride `<!-- -->` comments in the Markdown read view.
3. **Lossless view is `doced read --ast`** (JSON); the default `read` is Markdown with locator hints.
4. **`--batch FILE.jsonl`** applies many changes from one read (locators stay valid across the batch).
5. **`--dry-run` / `-o/--output`** preview or write a copy instead of in place.
6. **A zero-effect mutation is an error** (nonzero exit), so the LLM's done/retry decision off the exit code is reliable.
7. **Untrusted content guard:** the skill explicitly instructs the LLM that doc content is data, not instructions.
---
## 5. Command surface (`doced <verb>`)
Thin CLI surface; all domain logic lives in `src/doced/ops`. Every command has `--help`.
**Read / inspect**
- `doced read FILE [--ast | --json] [--from pN --to pM]` — Markdown (default) or lossless JSON AST.
- `doced find FILE "phrase"` — returns exact locator spans.
- `doced outline FILE` — heading tree as a locator tree.
- `doced wc FILE [LOCATOR]` — word/token count.
**Edit (in place)**
- `doced edit FILE --at LOCATOR --text "…"` — edit text at a locator.
- `doced insert FILE --after pN --text "…"` / `--at-start` / `--at-end`.
- `doced replace FILE "pattern" "replacement" [--all]` — text replace, run-format preserved.
- `doced delete FILE --at LOCATOR`.
**Legal-grade features (the differentiation)**
- `doced track-changes FILE on|off|list|accept|reject` — tracked changes / redlines.
- `doced comments FILE add|list|reply|resolve [--anchor PHRASE] [--at cN]`.
- `doced content-controls FILE list|add|edit` — document controls / structured document tags.
- `doced fields FILE list|update|set` — cross-reference, hyperlink, page-number, author, etc.
- `doced toc FILE add|update` — table of contents.
- `doced headers FILE set|list` / `doced footers FILE set|list [--page-number]`.
**Tables & media**
- `doced tables FILE create|insert-row|insert-col|set-cell|format`.
- `doced images FILE add|list|replace`.
**Authoring**
- `doced create OUT.docx --from draft.md` — Markdown (GFM) → docx.
- `doced create OUT.docx --text-file file.txt` — verbatim text channel (no parser).
**Deep / safety**
- `doced raw FILE part list|get --at NAME` — direct OPC part access (gated).
- `doced diff FILE --against OLD.docx` — structural diff vs a baseline.
- `doced validate FILE` — ECMA-376 schema check.
- `doced render FILE --out DIR` — rasterize pages (LibreOffice headless) when layout is the question.
**Meta**
- `doced --version`, `doced <verb> --help`, `doced info locators`, `doced info schema`.
---
## 6. Installation & packaging (Debian/Ubuntu)
Two paths, both drop a `doced` on PATH:
**A. pipx (primary, zero system pollution)**
```bash
pipx install doced
doced --version
```
`pyproject.toml` declares the `doced = "doced.cli:main"` console-script entry point. pipx gives each tool an isolated env — perfect for an agent's terminal.
**B. apt `.deb` (enterprise/immutable hosts)**
```bash
sudo dpkg -i dist/doced_0.1.0_amd64.deb
doced --version
```
`debian/` + `build-deb.sh` mirror the existing `docx-cli` recipe: binary→`/usr/local/bin/doced`, optional systemd service with sandboxing (`NoNewPrivileges`, `ProtectSystem=strict`, `PrivateTmp`).
**Install-time guarantees (from docx-cli's `install.sh`):**
- Pinned release + **SHA-256 verification** against published `SHA256SUMS`; never pipe a remote script into a shell.
- Platform detection (linux x64/arm64).
**Runtime dependency:** LibreOffice (`libreoffice`) optional, only for `doced render`. Everything else is pure Python + lxml.
---
## 7. Roadmap (phased)
| Phase | Status | Deliverable |
|---|---|---|
| **P0 — Scaffold** | pending | `pyproject.toml`, `doced` entry point, package I/O (zip + lxml), `--version`, `--help`, exit-code output module. |
| **P1 — Read layer** | pending | `read` (Markdown + `--ast`), `find`, `outline`, `wc`; locator model; Markdown read view with locator hints. |
| **P2 — Edit layer** | pending | `edit`, `insert`, `replace`, `delete`, `--batch`, `--dry-run`/`-o`; in-place mutation invariant; unit + LibreOffice round-trip fixtures. |
| **P3 — Legal features** | pending | `track-changes`, `comments`, `content-controls`, `fields`, `toc`, `headers`/`footers`. |
| **P4 — Tables/media/author** | pending | `tables`, `images`, `create` (Markdown + verbatim text). |
| **P5 — Deep/safety** | pending | `raw` (gated part access), `diff`, `validate`, `render`. |
| **P6 — Skill** | pending | `doced_skill.md` — LLM guide modeled on the existing skill (Python-flavored, legally oriented). |
| **P7 — Packaging** | pending | pipx install + `.deb` (`debian/`, `build-deb.sh`) + `install.sh` (pinned + SHA-256). |
| **P8 — Dockerize** | pending | Production Docker image (post host-verification), per deployment strategy. |
---
## 8. Testing discipline
- **Unit:** `lxml`-shape assertions on every op (proves the emitted XML is correct).
- **Fixture round-trip:** each new verb gets a builder fixture that round-trips through **LibreOffice headless** (catches render-invalidity: bad child order, broken parts) — the same discipline `docx-cli` enforces.
- **Weak-agent scenario:** each feature gets a task scenario proving an LLM can discover + use it (the gap unit tests structurally can't see).
- **Contract tests:** exit codes + one-line acks + `--batch` locator-stability.
- Run on the host venv; sub-agents execute tests in a single pass, report issues, then isolated sub-agents fix (per workspace testing discipline).
---
## 9. Open questions for the user
1. **Language:** Python (this plan) vs. reusing the existing Bun `docx-cli` as the engine and wrapping it? I recommend **Python + lxml** for stack alignment and legal-team maintainability, but confirm.
2. **Reuse vs. replace:** Should `doced` be a **fresh implementation**, or should we build `doced` as a **thin Python layer over** the existing `docx-cli` binary (reuse its fidelity engine)? Fresh Python gives independence; over `docx-cli` reuses a mature engine.
3. **Scope:** Is the full legal-feature set (§5) in scope for v1, or a trimmed MVP (read/find/edit + comments + tracked changes) first?
4. **Distribution priority:** pipx-first, .deb-first, or both from the start?
5. **Web research gap:** Web search is blocked. Should I try an alternative method (e.g., `fetch_url` on specific GitHub/pypa pages) to strengthen the competitive analysis, or proceed with the workspace-grounded analysis?
---
## 10. Immediate next step
On your go, I'll scaffold **P0** (pyproject, entry point, package I/O, `--version`/`--help`, exit-code module), verify it runs on the host venv, and commit on a feature branch.
+54
View File
@@ -0,0 +1,54 @@
# py-docx-mcp
Python MCP server for DOCX document manipulation, optimized for OpenWebUI (Streamable HTTP).
## Features
- Full-featured DOCX tools:
- Create, open, edit, merge, split documents
- Add paragraphs, headings, tables, lists, images, hyperlinks
- Find/replace (plain and regex)
- Template-based document generation
- Export to Markdown/HTML
- Metadata, structure, and search utilities
- Transport:
- OpenWebUI MCP (Streamable HTTP)
- Auth:
- Optional API key via environment (Bearer or X-API-Key)
- Document delivery:
- Generated files returned as base64 content to the caller (no external mounts required)
## Quick Start
### Build
- docker build -t py-docx-mcp .
### Run
- docker run --rm -p 3000:3000 py-docx-mcp
### With API key
- docker run --rm -p 3000:3000 -e DOCX_MCP_API_KEY="your-secret-key" py-docx-mcp
### Connect in OpenWebUI
- Admin Settings → External Tools → + Add Server
- Type: MCP (Streamable HTTP)
- URL: http://<host>:3000
- Auth:
- If DOCX_MCP_API_KEY is set: Bearer with your key
- Otherwise: None
## Environment
- DOCX_MCP_API_KEY: API key (optional but recommended)
- DOCX_MCP_HTTP_HOST: Bind host (default: 0.0.0.0)
- DOCX_MCP_HTTP_PORT: Bind port (default: 3000)
- DOCX_MCP_TEMPLATES_DIR: Templates directory (default: /templates)
- DOCX_MCP_MAX_SIZE: Max document size in bytes (default: 104857600)
- DOCX_MCP_MAX_DOCS: Max open documents (default: 30)
- DOCX_MCP_SANDBOX: Enable sandbox mode (default: true)
- DOCX_MCP_ALLOW_EXTERNAL_TOOLS: Allow external tools (default: false)
- DOCX_MCP_ALLOW_NETWORK: Allow network access (default: false)
+14 -38
View File
@@ -1,50 +1,26 @@
[build-system]
requires = ["setuptools>=68", "wheel"]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
[project]
name = "doced"
name = "py-docx-mcp"
version = "0.1.0"
description = "A terminal-native CLI that lets an LLM interact with DOCX files at an advanced level on Debian/Ubuntu."
readme = "README.md"
requires-python = ">=3.12"
license = { text = "MIT" }
authors = [{ name = "doced contributors" }]
keywords = ["docx", "word", "ooxml", "llm", "agent", "cli", "legal"]
classifiers = [
"Development Status :: 3 - Alpha",
"Environment :: Console",
"Intended Audience :: Developers",
"Intended Audience :: Legal Industry",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3.12",
"Topic :: Text Processing :: Markup",
]
description = "Python MCP server for DOCX document manipulation"
requires-python = ">=3.10"
dependencies = [
"lxml>=5.3.0,<6.0.0",
]
[project.optional-dependencies]
# Only needed for `doced render` (rasterize pages to PNG via LibreOffice).
render = []
# Development / test extras.
dev = [
"pytest>=8.0.0",
"pytest-cov>=5.0.0",
"fastapi>=0.115.0",
"uvicorn[standard]>=0.24.0",
"python-docx>=1.1.0",
"Pillow>=10.0.0",
"markdown>=3.5",
"html5lib>=1.1",
"regex>=2024.0.0",
"aiofiles>=24.0.0",
]
[project.scripts]
doced = "doced.cli:main"
[project.urls]
Homepage = "https://example.com/doced"
Source = "https://example.com/doced"
py-docx-mcp = "server:main"
[tool.setuptools.packages.find]
where = ["src"]
include = ["doced*"]
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
addopts = "-q"
include = ["*"]
+1
View File
@@ -0,0 +1 @@
# py-docx-mcp: Python MCP server for DOCX document manipulation
-13
View File
@@ -1,13 +0,0 @@
"""doced — a terminal-native CLI for advanced DOCX manipulation by an LLM.
The package treats a ``.docx`` as an OPC package (a zip of XML parts) and
mutates it *in place* via ``lxml`` — never by rebuilding from a lossy
round-trip. See :mod:`doced.package` for the package I/O and
:mod:`doced.xmlcore` for the XML helpers.
"""
from __future__ import annotations
__version__ = "0.1.0"
__all__ = ["__version__"]
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
-1
View File
@@ -1 +0,0 @@
"""AST model, locator model, and the reader (view builder)."""
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
-120
View File
@@ -1,120 +0,0 @@
"""The document model: package + AST + locator index, mutated in place.
This is the heart of the in-place-mutation model. A :class:`DocumentModel`
owns an open :class:`~doced.package.Package` and a parsed
:class:`~doced.ast.model.Document` AST. The AST is a **read-only view** — we
never mutate AST objects. Instead we keep a locator→``lxml`` node index
(:attr:`nodes`) and mutate the XML directly, then :meth:`rebuild` the AST.
Core invariant (from the docx-cli design):
> **Changes must survive the write→read loop.** Mutate ``lxml`` element
> refs in place; only emit fresh XML for inserted nodes. Unmodeled elements
> are preserved, not rewritten.
Because positional ids (``pN`` / ``tN`` / …) are assigned in document order
during a read, an id may shift after a *structural* edit (insert/delete). We
therefore rebuild the AST (and the index) after every mutation, so callers
always address the *current* document.
"""
from __future__ import annotations
from lxml import etree as ET
from .. import xmlcore as xc
from ..ast import reader
from ..ast.locator import Locator, parse_locator
from ..ast.model import Document
from ..package import Package
class DocumentModel:
"""An open document ready for read + in-place mutation."""
def __init__(self, pkg: Package):
self.pkg = pkg
self.ast: Document = reader.build_document(pkg)
self.nodes: dict[str, ET._Element] = {}
self._index()
# ------------------------------------------------------------------
# Indexing
# ------------------------------------------------------------------
def _index(self) -> None:
"""Build the locator → node index from the current XML.
Walks ``word/document.xml`` in document order and assigns positional
ids (``pN`` paragraphs, ``tN`` tables, ``sN`` sections) directly, so
the index never depends on the AST. Rebuild after any mutation.
"""
self.nodes = {}
root = self.pkg.parse("word/document.xml")
body = xc.find(root, "w:body")
if body is None:
body = root # degenerate document; treat root as body
# Comments live in a separate part.
if self.pkg.has_part("word/comments.xml"):
croot = self.pkg.parse("word/comments.xml")
for c in croot.findall(xn("w", "comment")):
cid = c.get(xn("w", "id")) or "c0"
self.nodes[cid] = c
p_idx = t_idx = s_idx = 0
for el in body:
tag = xc.local(el.tag)
if tag == "p":
self.nodes[f"p{p_idx}"] = el
p_idx += 1
elif tag == "tbl":
self.nodes[f"t{t_idx}"] = el
t_idx += 1
elif tag == "sectPr":
self.nodes[f"s{s_idx}"] = el
s_idx += 1
# ------------------------------------------------------------------
# Root element access
# ------------------------------------------------------------------
def root(self) -> ET._Element:
"""Return the root element of ``word/document.xml``."""
return self.pkg.parse("word/document.xml")
# ------------------------------------------------------------------
# Save
# ------------------------------------------------------------------
def save(self, path: str | None = None) -> None:
"""Write the (mutated) package to *path* (defaults to the source)."""
self.pkg.save(path)
def rebuild(self) -> None:
"""Rebuild the AST + locator index from the current XML."""
self.ast = reader.build_document(self.pkg)
self._index()
# ------------------------------------------------------------------
# Locator resolution
# ------------------------------------------------------------------
def resolve(self, locator: str | Locator) -> ET._Element:
"""Resolve a locator string/locator to its ``lxml`` node.
Raises:
KeyError: if the locator does not address any node.
"""
if isinstance(locator, str):
locator = parse_locator(locator)
key = self._locator_key(locator)
if key not in self.nodes:
raise KeyError(f"locator not found: {locator}")
return self.nodes[key]
def _locator_key(self, locator: Locator) -> str:
if locator.kind == "cell" and locator.container and locator.row is not None and locator.col is not None:
return f"{locator.container}:{locator.row}c{locator.col}"
return str(locator.id)
def xn(prefix: str, name: str) -> str:
"""Namespace-qualified tag/attr name."""
return f"{{{xc.NSMAP[prefix]}}}{name}"
-196
View File
@@ -1,196 +0,0 @@
"""The locator model: stable, positional addressing for an LLM.
Locators are how an LLM addresses content without re-reading the whole file.
They are **positional** (``p3`` = the 3rd paragraph in document order) and
**stable within a single read** — they shift only after *structural* edits
(insert/delete/section changes), never after in-place content/format edits.
Grammar (all parsed by :func:`parse_locator`):
.. code-block:: text
pN a paragraph (block)
pN:pM or pN-pM a range of paragraphs
pN:a-b a character span within paragraph pN
tN a table
tN:rR a row of table tN
tN:cC a column of table tN
tN:rRcC a cell (row R, column C) of table tN
tN:rRcC:pK a paragraph inside that cell
sN a section break
cN a comment
tcN a tracked change
imgN an image
linkN a hyperlink
fnN / enN a footnote / endnote
hdrN / ftrN a header / footer marginal
Locators ride ``<!-- ... -->`` HTML comments in the Markdown read view, so an
LLM sees the handle it needs and can feed it straight back into ``--at``.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from typing import Union
# ---------------------------------------------------------------------------
# Locator kinds (mirrors the AST node types they address).
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class Locator:
"""A parsed locator. ``kind`` selects the addressed node type."""
kind: str
# The primary id (e.g. "p3", "t0", "c1"). ``start``/``end`` are character
# offsets for ``blockSpan`` locators; row/col for cell locators.
id: str | None = None
start: int | None = None
end: int | None = None
row: int | None = None
col: int | None = None
# For cell-scoped locators, the container table id (e.g. "t0").
container: str | None = None
# For nested cell locators (paragraph inside a cell), the chained id.
inner: "Locator | None" = None
# Internal-only: the marginal kind, only meaningful for kind=="marginal".
_marginal_kind: str | None = None
def __str__(self) -> str:
return self._format()
def _format(self) -> str:
if self.kind == "comment":
return str(self.id)
if self.kind == "cell":
base = f"{self.container}:{self.row}c{self.col}"
return f"{base}:{self.inner}" if self.inner else base
if self.kind == "cellRange":
return f"{self.container}:{self.row}c{self.col}-{self.end}c{self.col}"
return str(self.id)
# ---------------------------------------------------------------------------
# Parser.
# ---------------------------------------------------------------------------
_BLOCK_RE = re.compile(r"^(p|t|s)(\d+)$")
_SPAN_RE = re.compile(r"^p(\d+):(\d+)-(\d+)$")
_BLOCK_RANGE_RE = re.compile(r"^p(\d+)-p(\d+)$")
_CELL_RE = re.compile(r"^(t\d+):r(\d+)c(\d+)(?::(.+))?$")
_CELL_RANGE_RE = re.compile(r"^(t\d+):r(\d+)c(\d+)-r(\d+)c(\d+)$")
_TABLE_ROW_RE = re.compile(r"^(t\d+):r(\d+)$")
_TABLE_COL_RE = re.compile(r"^(t\d+):c(\d+)$")
_COMMENT_RE = re.compile(r"^c(\d+)$")
_TRACKED_RE = re.compile(r"^tc(\d+)$")
_IMAGE_RE = re.compile(r"^img(\d+)$")
_LINK_RE = re.compile(r"^link(\d+)$")
_FOOTNOTE_RE = re.compile(r"^fn(\d+)$")
_ENDNOTE_RE = re.compile(r"^en(\d+)$")
_MARGINAL_RE = re.compile(r"^(hdr|ftr)(\d+)$")
class LocatorParseError(ValueError):
"""Raised when a locator string cannot be parsed."""
def parse_locator(input_str: str) -> Locator:
"""Parse a locator string into a :class:`Locator`.
Raises:
LocatorParseError: if the string is empty or unrecognizable.
"""
s = input_str.strip()
if not s:
raise LocatorParseError("empty locator")
# Cell range: tN:rRcC-rRcC
m = _CELL_RANGE_RE.match(s)
if m:
tid, r1, c1, r2, c2 = m.groups()
return Locator(
kind="cellRange",
container=tid,
row=int(r1),
col=int(c1),
end=int(r2),
)
# Cell: tN:rRcC[:inner]
m = _CELL_RE.match(s)
if m:
tid, r, c, rest = m.groups()
inner = parse_locator(rest) if rest else None
return Locator(kind="cell", container=tid, row=int(r), col=int(c), inner=inner)
# Table row: tN:rR
m = _TABLE_ROW_RE.match(s)
if m:
tid, r = m.groups()
return Locator(kind="tableRow", container=tid, row=int(r))
# Table column: tN:cC
m = _TABLE_COL_RE.match(s)
if m:
tid, c = m.groups()
return Locator(kind="tableColumn", container=tid, col=int(c))
# Comment / tracked change / image / link / notes / marginal
m = _COMMENT_RE.match(s)
if m:
return Locator(kind="comment", id=f"c{m.group(1)}")
m = _TRACKED_RE.match(s)
if m:
return Locator(kind="trackedChange", id=f"tc{m.group(1)}")
m = _IMAGE_RE.match(s)
if m:
return Locator(kind="image", id=f"img{m.group(1)}")
m = _LINK_RE.match(s)
if m:
return Locator(kind="hyperlink", id=f"link{m.group(1)}")
m = _FOOTNOTE_RE.match(s)
if m:
return Locator(kind="footnote", id=f"fn{m.group(1)}")
m = _ENDNOTE_RE.match(s)
if m:
return Locator(kind="endnote", id=f"en{m.group(1)}")
m = _MARGINAL_RE.match(s)
if m:
kind = "header" if m.group(1) == "hdr" else "footer"
return Locator(kind="marginal", id=f"{m.group(1)}{m.group(2)}", _marginal_kind=kind)
# Block range: pN-pM
m = _BLOCK_RANGE_RE.match(s)
if m:
a, b = m.groups()
if int(b) < int(a):
raise LocatorParseError("end paragraph precedes start")
return Locator(kind="blockRange", id=f"p{a}", end=b)
# Block span: pN:a-b
m = _SPAN_RE.match(s)
if m:
p, a, b = m.groups()
if int(b) < int(a):
raise LocatorParseError("end offset precedes start")
return Locator(kind="blockSpan", id=f"p{p}", start=int(a), end=int(b))
# Plain block: pN / tN / sN
m = _BLOCK_RE.match(s)
if m:
prefix, idx = m.groups()
kind = "block" if prefix == "p" else ("section" if prefix == "s" else "table")
return Locator(kind=kind, id=f"{prefix}{idx}")
raise LocatorParseError(f"unrecognized locator syntax: {input_str!r}")
def is_cell_scoped(input_str: str) -> bool:
"""True if a locator addresses content inside a table cell."""
return bool(re.search(r":r\d+c\d+(?::|$)", input_str.strip()))
def is_relationship_locator(input_str: str) -> bool:
"""True for an ``rIdN`` relationship handle."""
return bool(re.match(r"^rId\d+$", input_str.strip()))
-258
View File
@@ -1,258 +0,0 @@
"""The AST model: typed nodes describing a document's structure.
These dataclasses are the in-memory representation an LLM reads. They are a
*view* over the underlying XML — the source of truth is the OOXML in the
package, mutated in place. The AST gives an LLM a stable, structured,
JSON-serializable picture of that XML without forcing it to reason about raw
tags.
Nodes:
* :class:`Document` — the root; holds body blocks, properties, and side tables
(comments, notes, numbering, styles, marginals).
* :class:`Paragraph`, :class:`Run`, :class:`Table` / :class:`TableRow` /
:class:`TableCell` — the body.
* :class:`SectionBreak`, :class:`Marginal` — layout and page chrome.
* :class:`Comment`, :class:`Note`, :class:`TrackedChange` — legal-grade features.
The :class:`~doced.ast.doc.DocumentModel` owns the package + this AST and
performs in-place XML mutation.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Union
# ---------------------------------------------------------------------------
# Properties
# ---------------------------------------------------------------------------
@dataclass
class DocProperties:
"""Core document properties (``docProps/core.xml``)."""
title: str | None = None
author: str | None = None
created: str | None = None
modified: str | None = None
keywords: str | None = None
subject: str | None = None
# ---------------------------------------------------------------------------
# Runs (inline content within a paragraph)
# ---------------------------------------------------------------------------
@dataclass
class TextRun:
"""A run of text with character-level formatting."""
type: str = "text"
text: str = ""
bold: bool = False
italic: bool = False
underline: bool = False
strike: bool = False
color: str | None = None
font: str | None = None
size_half_points: int | None = None # 2 * point size
highlight: str | None = None
hyperlink_id: str | None = None
tracked_change_id: str | None = None
@dataclass
class BreakRun:
"""A line or page break."""
type: str = "break"
kind: str = "line" # "line" | "page" | "column"
tracked_change_id: str | None = None
@dataclass
class TabRun:
"""A tab character."""
type: str = "tab"
tracked_change_id: str | None = None
Run = Union[TextRun, BreakRun, TabRun]
# ---------------------------------------------------------------------------
# Paragraph
# ---------------------------------------------------------------------------
@dataclass
class Paragraph:
"""A paragraph (block). Carries its positional id ``pN``."""
id: str
type: str = "paragraph"
style: str | None = None
alignment: str | None = None # left | center | right | justify
runs: list[Run] = field(default_factory=list)
list_level: int | None = None
list_id: str | None = None
indent_left: int | None = None # twips
space_before: int | None = None # twips
space_after: int | None = None # twips
# Tracked change wrapping this paragraph (e.g. a deleted paragraph).
tracked_change_id: str | None = None
@property
def text(self) -> str:
"""The concatenated text of all text runs (for search/replace)."""
return "".join(r.text for r in self.runs if isinstance(r, TextRun))
# ---------------------------------------------------------------------------
# Tables
# ---------------------------------------------------------------------------
@dataclass
class TableCell:
"""A table cell, containing nested blocks."""
blocks: list[Union["Paragraph", "Table", "SectionBreak"]] = field(default_factory=list)
grid_span: int = 1
v_merge: str | None = None # restart | continue
shading: str | None = None
width_twips: int | None = None
v_align: str | None = None # top | center | bottom
@dataclass
class TableRow:
"""A table row."""
cells: list[TableCell] = field(default_factory=list)
repeat_header: bool = False
height_twips: int | None = None
tracked_change_id: str | None = None
@dataclass
class Table:
"""A table (block). Carries its positional id ``tN``."""
id: str
type: str = "table"
rows: list[TableRow] = field(default_factory=list)
grid: list[int] = field(default_factory=list) # column widths in twips
style: str | None = None
align: str | None = None # left | center | right
# ---------------------------------------------------------------------------
# Sections & marginals
# ---------------------------------------------------------------------------
@dataclass
class SectionBreak:
"""A section break. Carries its positional id ``sN``."""
id: str
type: str = "sectionBreak"
columns: int | None = None
section_type: str | None = None # continuous | nextPage | evenPage | oddPage
page_orientation: str | None = None # portrait | landscape
margin_top: int | None = None # twips
margin_right: int | None = None
margin_bottom: int | None = None
margin_left: int | None = None
@dataclass
class Marginal:
"""A header or footer. Carries its positional id (``hdrN`` / ``ftrN``)."""
id: str
kind: str # header | footer
type: str = "default" # default | first | even
text: str = ""
section_id: str | None = None
# ---------------------------------------------------------------------------
# Legal-grade features
# ---------------------------------------------------------------------------
@dataclass
class Comment:
"""A document comment. Carries its positional id ``cN``."""
id: str
author: str = ""
text: str = ""
date: str | None = None
parent_id: str | None = None
resolved: bool = False
start_block: str | None = None
start_offset: int = 0
end_block: str | None = None
end_offset: int = 0
@dataclass
class Note:
"""A footnote or endnote body. Carries its positional id (``fnN`` / ``enN``)."""
id: str
kind: str # footnote | endnote
text: str = ""
@dataclass
class TrackedChange:
"""A tracked change (insertion / deletion / move). Id ``tcN``."""
id: str
kind: str # ins | del | moveFrom | moveTo
author: str = ""
date: str | None = None
revision_id: str = ""
paragraph_id: str | None = None
# ---------------------------------------------------------------------------
# Document
# ---------------------------------------------------------------------------
Block = Union[Paragraph, Table, SectionBreak]
@dataclass
class Document:
"""The root AST node: body blocks plus side tables."""
type: str = "document"
blocks: list[Block] = field(default_factory=list)
properties: DocProperties = field(default_factory=DocProperties)
comments: list[Comment] = field(default_factory=list)
footnotes: list[Note] = field(default_factory=list)
endnotes: list[Note] = field(default_factory=list)
tracked_changes: list[TrackedChange] = field(default_factory=list)
# Numbering definitions (abstractNum / num) — surfaced on demand.
styles: dict[str, dict] = field(default_factory=dict)
# Marginals keyed by section id -> {"header": [...], "footer": [...]}.
marginals: dict[str, dict] = field(default_factory=dict)
path: str | None = None
# ------------------------------------------------------------------
# Convenience accessors
# ------------------------------------------------------------------
def paragraphs(self) -> list[Paragraph]:
return [b for b in self.blocks if isinstance(b, Paragraph)]
def tables(self) -> list[Table]:
return [b for b in self.blocks if isinstance(b, Table)]
def sections(self) -> list[SectionBreak]:
return [b for b in self.blocks if isinstance(b, SectionBreak)]
def get_block(self, block_id: str) -> Block | None:
for b in self.blocks:
if b.id == block_id:
return b
return None
def all_notes(self) -> list[Note]:
return self.footnotes + self.endnotes
-315
View File
@@ -1,315 +0,0 @@
"""The reader: parse ``word/document.xml`` into the typed AST.
This is the *view* builder — it walks the parsed XML tree and populates a
:class:`~doced.ast.model.Document`. It never mutates; it only reads. The
positional ids (``pN`` / ``tN`` / ``sN`` / ``cN`` / ``tcN`` / …) are assigned
in document order here, so they are stable within a single read and shift only
after structural edits (see :mod:`doced.ast.locator`).
The reader is namespace-aware (see :mod:`doced.xmlcore`) and tolerant: unknown
elements are skipped rather than raising, so a document with features we don't
model still reads.
"""
from __future__ import annotations
from lxml import etree as ET
from .. import xmlcore as xc
from ..package import Package, PackagePartMissing
from . import model as M
def _local(el: ET._Element) -> str:
return xc.local(el.tag)
def _xn(prefix: str, name: str) -> str:
"""Namespace-qualified attribute name, e.g. ``{w}val``."""
return f"{{{xc.NSMAP[prefix]}}}{name}"
def _opt(el: ET._Element | None, prefix: str, name: str, attr: str) -> str | None:
"""Return ``el.{prefix}:{attr}`` if ``el`` and the attr exist, else None."""
if el is None:
return None
val = el.get(_xn(prefix, attr))
return val if val else None
# ---------------------------------------------------------------------------
# Character formatting extraction.
# ---------------------------------------------------------------------------
def _run_format(rPr: ET._Element | None) -> dict:
"""Extract character formatting from a ``<w:rPr>`` element."""
fmt: dict = {}
if rPr is None:
return fmt
if xc.find(rPr, "w:b") is not None:
fmt["bold"] = _opt(xc.find(rPr, "w:b"), "w", "b", "val") != "false"
if xc.find(rPr, "w:i") is not None:
fmt["italic"] = True
u = xc.find(rPr, "w:u")
if u is not None and u.get(_xn("w", "val")) != "none":
fmt["underline"] = True
if xc.find(rPr, "w:strike") is not None:
fmt["strike"] = True
color = xc.find(rPr, "w:color")
col = _opt(color, "w", "color", "val")
if col and col != "auto":
fmt["color"] = col
font = xc.find(rPr, "w:rFonts")
f = _opt(font, "w", "rFonts", "ascii")
if f:
fmt["font"] = f
sz = xc.find(rPr, "w:sz")
sz_val = _opt(sz, "w", "sz", "val")
if sz_val:
fmt["size_half_points"] = int(sz_val)
return fmt
# ---------------------------------------------------------------------------
# Paragraph reading.
# ---------------------------------------------------------------------------
def _paragraph_runs(pEl: ET._Element) -> tuple[list[M.Run], str | None]:
"""Extract runs and the paragraph style id from a ``<w:p>`` element."""
runs: list[M.Run] = []
style = _opt(xc.find(pEl, "w:pPr"), "w", "pStyle", "val")
for child in pEl:
tag = _local(child)
if tag == "r":
run = _run_element(child)
if run is not None:
runs.append(run)
elif tag == "br":
br_type = _opt(child, "w", "br", "type")
kind = "line"
if br_type == "page":
kind = "page"
elif br_type == "column":
kind = "column"
runs.append(M.BreakRun(kind=kind))
elif tag == "tab":
runs.append(M.TabRun())
return runs, style
def _run_element(rEl: ET._Element) -> M.Run | None:
"""Build a run from a ``<w:r>`` element."""
t = xc.find(rEl, "w:t")
if t is not None and t.text:
fmt = _run_format(xc.find(rEl, "w:rPr"))
hyperlink = xc.find(rEl, "r:hyperlink")
hl_id = _opt(hyperlink, "r", "hyperlink", "id")
return M.TextRun(
text=t.text,
bold=fmt.get("bold", False),
italic=fmt.get("italic", False),
underline=fmt.get("underline", False),
strike=fmt.get("strike", False),
color=fmt.get("color"),
font=fmt.get("font"),
size_half_points=fmt.get("size_half_points"),
hyperlink_id=hl_id,
)
return None
def _paragraph_alignment(pEl: ET._Element) -> str | None:
return _opt(xc.find(pEl, "w:pPr"), "w", "jc", "val")
def _paragraph_list(pEl: ET._Element) -> tuple[int | None, str | None]:
numPr = xc.find(pEl, "w:pPr/w:numPr")
if numPr is None:
return None, None
level = _opt(numPr, "w", "ilvl", "val")
nid = _opt(numPr, "w", "numId", "val")
return int(level) if level else None, nid
# ---------------------------------------------------------------------------
# Table reading.
# ---------------------------------------------------------------------------
def _table_grid(tbl: ET._Element) -> list[int]:
widths: list[int] = []
grid = xc.find(tbl, "w:tblGrid")
if grid is not None:
for col in grid.findall(xn("w", "gridCol")):
w = col.get(_xn("w", "w"))
if w:
widths.append(int(w))
return widths
def _table_cell(tc: ET._Element) -> M.TableCell:
cell = M.TableCell()
tcPr = xc.find(tc, "w:tcPr")
if tcPr is not None:
span = xc.find(tcPr, "w:gridSpan")
if span is not None and span.get(_xn("w", "val")):
cell.grid_span = int(span.get(_xn("w", "val")))
vMerge = xc.find(tcPr, "w:vMerge")
if vMerge is not None:
val = vMerge.get(_xn("w", "val"))
cell.v_merge = "restart" if val == "restart" else "continue"
shading = xc.find(tcPr, "w:shd")
if shading is not None and shading.get(_xn("w", "val")) not in (None, "clear"):
cell.shading = shading.get(_xn("w", "fill"))
tcW = xc.find(tcPr, "w:tcW")
if tcW is not None and tcW.get(_xn("w", "w")):
cell.width_twips = int(tcW.get(_xn("w", "w")))
vAlign = xc.find(tcPr, "w:vAlign")
if vAlign is not None:
cell.v_align = vAlign.get(_xn("w", "val"))
for child in tc:
if _local(child) == "p":
runs, style = _paragraph_runs(child)
para = M.Paragraph(id="", style=style, runs=runs)
cell.blocks.append(para)
return cell
def _read_table(tbl: ET._Element, idx: int) -> M.Table:
return M.Table(
id=f"t{idx}",
rows=[M.TableRow(cells=(_table_cell(tc) for tc in tbl.findall(_xn("w", "tr"))))],
grid=_table_grid(tbl),
)
# ---------------------------------------------------------------------------
# Section break reading.
# ---------------------------------------------------------------------------
def _section_break(sid: str, sectPr: ET._Element) -> M.SectionBreak:
sb = M.SectionBreak(id=sid)
for child in sectPr:
tag = _local(child)
if tag == "cols" and child.get(_xn("w", "num")):
sb.columns = int(child.get(_xn("w", "num")))
elif tag == "pgSz":
w = int(child.get(_xn("w", "w"), "0"))
h = int(child.get(_xn("w", "h"), "0"))
sb.page_orientation = "landscape" if w > h else "portrait"
elif tag == "pgMar":
sb.margin_top = _twips(child, "top")
sb.margin_right = _twips(child, "right")
sb.margin_bottom = _twips(child, "bottom")
sb.margin_left = _twips(child, "left")
return sb
def _twips(el: ET._Element, name: str) -> int | None:
val = el.get(_xn("w", name))
return int(val) if val else None
# ---------------------------------------------------------------------------
# Side parts.
# ---------------------------------------------------------------------------
def _read_comments(pkg: Package) -> list[M.Comment]:
if "word/comments.xml" not in pkg.parts:
return []
root = pkg.parse("word/comments.xml")
comments: list[M.Comment] = []
for c in root.findall(xn("w", "comment")):
cid = c.get(_xn("w", "id")) or f"c{len(comments)}"
author = (xc.find(c, "w:author") or ET.Element("x")).get(_xn("w", "val"), "")
date = (xc.find(c, "w:date") or ET.Element("x")).get(_xn("w", "val"))
text = "".join(t.text or "" for t in c.iter(xn("w", "t")))
comments.append(M.Comment(id=cid, author=author, date=date, text=text))
return comments
def _read_notes(pkg: Package, part: str, kind: str) -> list[M.Note]:
if part not in pkg.parts:
return []
root = pkg.parse(part)
notes: list[M.Note] = []
for n in root.findall(xn("w", kind)):
nid = n.get(_xn("w", "id")) or f"{kind[0]}{len(notes) + 1}"
notes.append(M.Note(id=nid, kind=kind, text=n.text or ""))
return notes
def _read_properties(pkg: Package) -> M.DocProperties:
if "docProps/core.xml" not in pkg.parts:
return M.DocProperties()
root = pkg.parse("docProps/core.xml")
props = M.DocProperties()
for tag, attr in (
("dc:title", "title"),
("dc:creator", "author"),
("dcterms:created", "created"),
("dcterms:modified", "modified"),
("dc:subject", "subject"),
("dc:keywords", "keywords"),
):
el = xc.find(root, tag)
if el is not None and el.text:
setattr(props, attr, el.text)
return props
# ---------------------------------------------------------------------------
# Document reading.
# ---------------------------------------------------------------------------
def _iter_body_elements(body: ET._Element) -> list[ET._Element]:
"""Return body elements in order, separating ``<w:sectPr>`` out."""
els: list[ET._Element] = []
sect_pr: ET._Element | None = None
for child in body:
if _local(child) == "sectPr":
sect_pr = child
else:
els.append(child)
return els, sect_pr
def build_document(pkg: Package) -> M.Document:
"""Parse the package into a typed :class:`M.Document`.
Raises:
PackagePartMissing: if ``word/document.xml`` is absent.
"""
if "word/document.xml" not in pkg.parts:
raise PackagePartMissing("word/document.xml not found in package")
root = pkg.parse("word/document.xml")
body = xc.find(root, "w:body")
if body is None:
body = root # degenerate document; treat root as body
doc = M.Document(path=pkg.path)
p_idx, t_idx, s_idx = 0, 0, 0
for el in body:
tag = _local(el)
if tag == "p":
runs, style = _paragraph_runs(el)
doc.blocks.append(M.Paragraph(
id=f"p{p_idx}",
style=style,
runs=runs,
alignment=_paragraph_alignment(el),
list_level=_paragraph_list(el)[0],
list_id=_paragraph_list(el)[1],
))
p_idx += 1
elif tag == "tbl":
doc.blocks.append(_read_table(el, t_idx))
t_idx += 1
# The trailing sectPr defines document-wide section properties.
sect_pr = xc.find(body, "w:sectPr")
if sect_pr is not None:
doc.blocks.append(_section_break(f"s{s_idx}", sect_pr))
doc.comments = _read_comments(pkg)
doc.footnotes = _read_notes(pkg, "word/footnotes.xml", "footnote")
doc.endnotes = _read_notes(pkg, "word/endnotes.xml", "endnote")
doc.properties = _read_properties(pkg)
return doc
-133
View File
@@ -1,133 +0,0 @@
"""CLI entry point and command dispatch.
``doced`` is a thin shell over :mod:`doced.ops` and :mod:`doced.features`.
Each command is a *command function* ``run(args: list[str]) -> int`` that
returns an exit code. Commands register themselves via :func:`register`, so
the dispatch table stays declarative and new verbs can be added by dropping a
module into :mod:`doced.ops` / :mod:`doced.features` and adding one line here.
The surface is deliberately small and uniform:
.. code-block:: text
doced --version
doced <command> --help
doced <command> FILE [args]
Every command prints a one-line ack (or a machine-parseable error) and returns
an exit code — see :mod:`doced.output` for the contract.
"""
from __future__ import annotations
import argparse
import sys
from typing import Callable, NamedTuple
from . import __version__
from . import output
# A command function takes normalized args (after the command name) and
# returns an exit code.
CommandFn = Callable[[list[str]], int]
# Registry: name -> (summary, command_fn)
_REGISTRY: dict[str, tuple[str, CommandFn]] = {}
def register(name: str, summary: str, fn: CommandFn) -> None:
"""Register a command. ``name`` is the kebab-case verb on the CLI."""
_REGISTRY[name] = (summary, fn)
class _Command(NamedTuple):
name: str
summary: str
fn: CommandFn
def _commands() -> list[_Command]:
return [_Command(n, s, f) for n, (s, f) in sorted(_REGISTRY.items())]
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="doced",
description="An LLM-driven CLI for advanced .docx manipulation.",
add_help=False,
)
parser.add_argument("--version", action="store_true", help="show version and exit")
parser.add_argument(
"-h", "--help", action="store_true", help="show this help and exit"
)
parser.add_argument("command", nargs="?", help="command to run")
parser.add_argument("args", nargs=argparse.REMAINDER)
return parser
def _print_version() -> int:
print(f"doced {__version__}")
return output.EXIT_OK
def _print_help() -> int:
lines = [
f"doced {__version__} — an LLM-driven CLI for advanced .docx manipulation.",
"",
"USAGE:",
" doced <command> [options] FILE [args]",
"",
"COMMANDS:",
]
for cmd in _commands():
lines.append(f" {cmd.name:<16} {cmd.summary}")
lines += [
"",
"GLOBAL OPTIONS:",
" --version show version and exit",
" -h, --help show this help and exit",
"",
"Run 'doced <command> --help' for command-specific options.",
"",
"The exit code is the success signal: 0 ok, 1 error, 2 usage, 3 not-found.",
]
print("\n".join(lines))
return output.EXIT_OK
def main(argv: list[str] | None = None) -> int:
"""CLI entry point. Returns an exit code (never raises for user errors)."""
argv = list(sys.argv[1:] if argv is None else argv)
parser = _build_parser()
try:
ns = parser.parse_args(argv)
except SystemExit as exc: # argparse already printed a message
return int(exc.code) if exc.code is not None else output.EXIT_USAGE
if ns.version:
return _print_version()
if ns.help or ns.command is None:
return _print_help()
fn = _REGISTRY.get(ns.command)
if fn is None:
return output.emit_error(
output.ErrorCode.USAGE,
f"unknown command: {ns.command}",
hint='Run "doced --help" for the list of commands.',
)
_, command_fn = fn
try:
return command_fn(ns.args)
except KeyboardInterrupt:
return output.EXIT_ERROR
except Exception as exc: # last-resort guard; commands should catch their own
return output.emit_error(
output.ErrorCode.UNHANDLED,
f"{type(exc).__name__}: {exc}",
)
if __name__ == "__main__":
sys.exit(main())
-1
View File
@@ -1 +0,0 @@
"""Command-line interface entry points."""
-1
View File
@@ -1 +0,0 @@
"""High-level feature operations (tracked changes, comments, tables, …)."""
-1
View File
@@ -1 +0,0 @@
"""In-place mutation operations for DOCX parts."""
Binary file not shown.
Binary file not shown.
-38
View File
@@ -1,38 +0,0 @@
"""Document loading: open a ``.docx`` package and parse it into a model.
This is the single choke point for reading a document. It opens the OPC
package and returns a :class:`~doced.ast.doc.DocumentModel` — the package plus
a parsed AST plus a locator index, ready for read + in-place mutation.
"""
from __future__ import annotations
from pathlib import Path
from ..ast.doc import DocumentModel
from ..package import (
NotAZipError,
Package,
PackageNotFound,
PackagePartMissing,
PartNotFoundError,
)
def open_document(path: str | Path) -> DocumentModel:
"""Open *path* and return a :class:`~doced.ast.doc.DocumentModel`.
Raises the package-level errors through so the CLI can map them to the
LLM contract's error codes.
"""
pkg = Package.open(path)
return DocumentModel(pkg)
__all__ = [
"open_document",
"PackageNotFound",
"PartNotFoundError",
"PackagePartMissing",
"NotAZipError",
]
-151
View File
@@ -1,151 +0,0 @@
"""The LLM contract: machine-readable output + an exit-code success signal.
This is the interface an LLM observes. It is modeled on the proven design used
by mature agent-facing DOCX tools:
1. **Exit code is the success signal.** ``0`` ok · ``1`` error · ``2`` usage ·
``3`` not-found. Every command prints a one-line text confirmation, so an
LLM never has to re-read a file just to know a mutation landed.
2. **A zero-effect mutation is an error** (nonzero exit), so the LLM's
done/retry decision off the exit code is reliable.
3. **Errors are machine-parseable**: ``{"code": "...", "error": "...",
"hint": "..."}`` — the LLM reads the ``code`` to decide recovery.
The one-line ack is text-first (not JSON) by default, so a weak model reading
stdout sees a human/LLM-friendly sentence; ``--verbose`` prints full JSON.
"""
from __future__ import annotations
import sys
from typing import Any
# ---------------------------------------------------------------------------
# Exit codes.
# ---------------------------------------------------------------------------
EXIT_OK = 0
EXIT_ERROR = 1
EXIT_USAGE = 2
EXIT_NOT_FOUND = 3
class ExitCode(int):
"""A typed exit-code sentinel (subclass of ``int`` for CLI ergonomics)."""
OK = EXIT_OK
ERROR = EXIT_ERROR
USAGE = EXIT_USAGE
NOT_FOUND = EXIT_NOT_FOUND
# ---------------------------------------------------------------------------
# Error codes (machine-readable ``code`` field on failures).
# ---------------------------------------------------------------------------
class ErrorCode:
USAGE = "USAGE"
FILE_NOT_FOUND = "FILE_NOT_FOUND"
PART_NOT_FOUND = "PART_NOT_FOUND"
NOT_A_ZIP = "NOT_A_ZIP"
INVALID_LOCATOR = "INVALID_LOCATOR"
BLOCK_NOT_FOUND = "BLOCK_NOT_FOUND"
RANGE_NOT_FOUND = "RANGE_NOT_FOUND"
MATCH_NOT_FOUND = "MATCH_NOT_FOUND"
COMMENT_NOT_FOUND = "COMMENT_NOT_FOUND"
IMAGE_NOT_FOUND = "IMAGE_NOT_FOUND"
IMAGE_SOURCE = "IMAGE_SOURCE"
HYPERLINK_NOT_FOUND = "HYPERLINK_NOT_FOUND"
RELATIONSHIP_NOT_FOUND = "RELATIONSHIP_NOT_FOUND"
TRACKED_CHANGE_NOT_FOUND = "TRACKED_CHANGE_NOT_FOUND"
TABLE_STRUCTURE = "TABLE_STRUCTURE"
INVALID_XML = "INVALID_XML"
VALIDATION_FAILED = "VALIDATION_FAILED"
RENDER_ENGINE = "RENDER_ENGINE"
RENDER_FAILED = "RENDER_FAILED"
LOCKED_FILE = "LOCKED_FILE"
ZERO_EFFECT = "ZERO_EFFECT"
UNHANDLED = "UNHANDLED"
# Exit code for each error code.
_EXIT_FOR_CODE: dict[str, int] = {
ErrorCode.USAGE: EXIT_USAGE,
ErrorCode.INVALID_LOCATOR: EXIT_USAGE,
ErrorCode.INVALID_XML: EXIT_USAGE,
ErrorCode.FILE_NOT_FOUND: EXIT_NOT_FOUND,
ErrorCode.PART_NOT_FOUND: EXIT_NOT_FOUND,
ErrorCode.BLOCK_NOT_FOUND: EXIT_NOT_FOUND,
ErrorCode.RANGE_NOT_FOUND: EXIT_NOT_FOUND,
ErrorCode.MATCH_NOT_FOUND: EXIT_NOT_FOUND,
ErrorCode.COMMENT_NOT_FOUND: EXIT_NOT_FOUND,
ErrorCode.IMAGE_NOT_FOUND: EXIT_NOT_FOUND,
ErrorCode.HYPERLINK_NOT_FOUND: EXIT_NOT_FOUND,
ErrorCode.RELATIONSHIP_NOT_FOUND: EXIT_NOT_FOUND,
ErrorCode.TRACKED_CHANGE_NOT_FOUND: EXIT_NOT_FOUND,
ErrorCode.NOT_A_ZIP: EXIT_ERROR,
ErrorCode.TABLE_STRUCTURE: EXIT_ERROR,
ErrorCode.VALIDATION_FAILED: EXIT_ERROR,
ErrorCode.IMAGE_SOURCE: EXIT_ERROR,
ErrorCode.RENDER_ENGINE: EXIT_ERROR,
ErrorCode.RENDER_FAILED: EXIT_ERROR,
ErrorCode.LOCKED_FILE: EXIT_ERROR,
ErrorCode.ZERO_EFFECT: EXIT_ERROR,
ErrorCode.UNHANDLED: EXIT_ERROR,
}
# ---------------------------------------------------------------------------
# Output helpers.
# ---------------------------------------------------------------------------
def emit_success(message: str, *, verbose: bool = False, payload: dict[str, Any] | None = None) -> int:
"""Print a success ack and return ``EXIT_OK``.
Args:
message: The one-line, text-first confirmation.
verbose: When true, also emit the full JSON payload.
payload: Optional structured data (e.g. minted locators).
"""
print(message)
if verbose and payload:
import json
print(json.dumps(payload))
return EXIT_OK
def emit_error(
code: str,
message: str,
*,
hint: str | None = None,
) -> int:
"""Print a machine-parseable error and return the mapped exit code.
The stdout line is ``{"code": "...", "error": "...", "hint": "..."}`` so an
LLM can parse the ``code`` to decide recovery. The hint is a human/LLM
friendly recovery suggestion.
"""
import json
payload: dict[str, Any] = {"code": code, "error": message}
if hint:
payload["hint"] = hint
print(json.dumps(payload))
return _EXIT_FOR_CODE.get(code, EXIT_ERROR)
def emit_json(payload: dict[str, Any]) -> int:
"""Print a full JSON payload (used by ``read --ast`` / ``--verbose``)."""
import json
print(json.dumps(payload))
return EXIT_OK
# The canonical "locator ids shift after structural edits" hint.
STALE_LOCATOR_HINT = (
"Locator ids are positional and shift after structural edits "
"(insert/delete/section changes) — an id from an earlier read may now point "
"elsewhere or be gone. Re-read the file to get current ids, then retry. "
"Applying several changes? Do them from ONE read with --batch so ids never "
"go stale mid-run."
)
-189
View File
@@ -1,189 +0,0 @@
"""OPC package I/O: open a ``.docx`` as a zip of XML parts.
A ``.docx`` is an Open Packaging Convention (OPC) package — a zip file whose
entries are *parts* (XML documents) plus media. :class:`Package` opens that
zip, reads parts into memory, and — crucially — **mutates parts in place** and
re-zips on save. We never rebuild a part from a lossy round-trip.
The class is deliberately thin: it owns bytes and part paths. Semantic views
(building the AST, resolving locators) live in :mod:`doced.ast`.
"""
from __future__ import annotations
import zipfile
from dataclasses import dataclass, field
from pathlib import Path
from . import xmlcore as xc
# Required / optional OPC part paths. ``word/document.xml`` is mandatory; the
# rest are optional and materialized lazily.
DOCUMENT_PART = "word/document.xml"
DEFAULT_CONTENT_TYPES = "[Content_Types].xml"
class DocedError(Exception):
"""Base class for all doced errors."""
class PackageNotFound(DocedError):
"""Raised when the target path does not exist."""
class NotAZipError(DocedError):
"""Raised when the file is not a valid zip / OPC package."""
class PartNotFoundError(DocedError):
"""Raised when a requested part is absent from the package."""
class PackagePartMissing(DocedError):
"""Raised when a required part (e.g. word/document.xml) is absent."""
@dataclass
class Package:
"""An in-memory OPC package.
Attributes:
path: The on-disk path this package was opened from (may be ``None``
for a freshly-created, never-saved package).
parts: Mapping of part path -> raw XML bytes.
media: Mapping of media path (e.g. ``word/media/img1.png``) -> bytes.
"""
path: str | None
parts: dict[str, bytes] = field(default_factory=dict)
media: dict[str, bytes] = field(default_factory=dict)
# ------------------------------------------------------------------
# Construction
# ------------------------------------------------------------------
@classmethod
def open(cls, path: str | Path) -> "Package":
"""Open an existing ``.docx`` from *path*.
Raises:
PackageNotFound: if the path does not exist.
NotAZipError: if the file is not a readable zip.
"""
p = Path(path)
if not p.exists():
raise PackageNotFound(f"file not found: {p}")
if not p.is_file():
raise NotAZipError(f"not a file: {p}")
try:
with zipfile.ZipFile(p, "r") as zf:
bad = zf.testzip()
if bad is not None:
raise NotAZipError(f"corrupt zip entry: {bad}")
parts = {
n: zf.read(n)
for n in zf.namelist()
if not n.startswith("word/media/")
}
media = {
n: zf.read(n)
for n in zf.namelist()
if n.startswith("word/media/")
}
except zipfile.BadZipFile as exc:
raise NotAZipError(f"not a valid zip/OPC package: {exc}") from exc
return cls(path=str(p), parts=parts, media=media)
@classmethod
def empty(cls, path: str | None = None) -> "Package":
"""Create an empty (or lazily-populated) package.
Used for authoring new documents and for ``--dry-run`` previews. Parts
are added on demand via :meth:`write_part`.
"""
return cls(path=path, parts={}, media={})
# ------------------------------------------------------------------
# Part access
# ------------------------------------------------------------------
def has_part(self, part: str) -> bool:
return part in self.parts
def read_part(self, part: str) -> bytes:
"""Return the raw bytes of *part*.
Raises:
PartNotFoundError: if the part is absent.
"""
if part not in self.parts:
raise PartNotFoundError(f"part not found: {part}")
return self.parts[part]
def read_text(self, part: str) -> str:
"""Return the decoded text of *part*."""
return self.read_part(part).decode("utf-8")
def parse(self, part: str):
"""Parse *part* into an lxml root element."""
return xc.parse_root(self.read_text(part))
def write_part(self, part: str, xml: str | bytes) -> None:
"""Stash *xml* for *part*; written to disk on :meth:`save`."""
self.parts[part] = xml.encode("utf-8") if isinstance(xml, str) else xml
def delete_part(self, part: str) -> None:
self.parts.pop(part, None)
# ------------------------------------------------------------------
# Media
# ------------------------------------------------------------------
def write_media(self, path: str, data: bytes) -> None:
self.media[path] = data
def read_media(self, path: str) -> bytes:
if path not in self.media:
raise PartNotFoundError(f"media not found: {path}")
return self.media[path]
# ------------------------------------------------------------------
# Save
# ------------------------------------------------------------------
def save(self, path: str | Path | None = None) -> None:
"""Re-zip the package and write it to *path* (defaults to the source).
Writes an initial ``[Content_Types].xml`` if the package was created
empty and none was provided.
"""
target = Path(path) if path else (
Path(self.path) if self.path else None
)
if target is None:
raise DocedError("cannot save: no path specified")
# Ensure a content-types part exists for a fresh package.
if DEFAULT_CONTENT_TYPES not in self.parts:
self.parts[DEFAULT_CONTENT_TYPES] = self._default_content_types_xml()
with zipfile.ZipFile(
target, "w", compression=zipfile.ZIP_DEFLATED
) as zf:
# Content types first (convention, not required, but tidy).
zf.writestr(DEFAULT_CONTENT_TYPES, self.parts[DEFAULT_CONTENT_TYPES])
for part, data in self.parts.items():
if part == DEFAULT_CONTENT_TYPES:
continue
zf.writestr(part, data)
for mpath, data in self.media.items():
zf.writestr(mpath, data)
@staticmethod
def _default_content_types_xml() -> str:
return (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>\n'
'<Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types">'
'<Default Extension="rels" ContentType="application/" />'
'<Default Extension="xml" ContentType="application/xml" />'
'<Override PartName="/word/document.xml" '
'ContentType="application/vnd.openxmlformats-officedocument.'
'wordprocessingml.document.main+xml" />'
"</Types>"
)
-131
View File
@@ -1,131 +0,0 @@
"""XML core helpers for DOCX (OOXML) processing.
All XML handling funnels through :mod:`doced.xmlcore` so that:
* parsing is **secure** (no external-entity / network access —
:data:`lxml.etree.default_parser` is configured with ``no_network`` and
``resolve_none_uri``),
* namespaces are handled through a single canonical map, and
* serialization preserves unmodeled elements byte-for-byte (in-place
mutation is the whole point — we never re-emit untouched subtrees).
"""
from __future__ import annotations
from lxml import etree as ET
# ---------------------------------------------------------------------------
# Canonical OOXML namespace map.
# ---------------------------------------------------------------------------
NSMAP: dict[str, str] = {
"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main",
"r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships",
"mc": "http://schemas.openxmlformats.org/markup-compatibility/2006",
"o": "http://schemas.openxmlformats.org/officeDocument/2006/officeMetadata",
"v": "urn:schemas-microsoft-com:vmo",
"w10": "urn:schemas-microsoft-com:office:word",
"wp": "http://schemas.openxmlformats.org/drawingml/2006/main",
"a": "http://schemas.openxmlformats.org/drawingml/2006/main",
"rId": "http://schemas.openxmlformats.org/officeDocument/2006/relationships",
"m": "http://schemas.openxmlformats.org/officeDocument/2006/math",
"w14": "http://schemas.microsoft.com/office/word/2010/wordml",
"w15": "http://schemas.microsoft.com/office/word/2012/wordml",
"w16": "http://schemas.microsoft.com/office/word/2018/wordml",
"w16c": "http://schemas.microsoft.com/office/word/2018/wordml/collection",
"w16e": "http://schemas.microsoft.com/office/word/2019/wordml",
"dc": "http://purl.org/dc/elements/1.1/",
"dcterms": "http://purl.org/dc/terms/",
"cp": "http://schemas.openxmlformats.org/package/2006/metadata/core-properties",
"xsi": "http://www.w3.org/2001/XMLSchema-instance",
}
# Element local-name lookup helpers.
def local(tag: str) -> str:
"""Return the local name of a namespaced ``{uri}local`` tag."""
if "}" in tag:
return tag.split("}", 1)[1]
return tag
def qn(prefix: str, name: str) -> str:
"""Build a fully-qualified ``{uri}name`` tag from the canonical map."""
uri = NSMAP.get(prefix)
if uri is None:
raise KeyError(f"unknown namespace prefix: {prefix!r}")
return f"{{{uri}}}{name}"
# ---------------------------------------------------------------------------
# Secure parser / serializer configuration.
# ---------------------------------------------------------------------------
def get_parser() -> ET.XMLParser:
"""Return a secure parser: no network, no external entities, DTD-safe.
``no_network`` blocks DTD/XSD fetches and ``resolve_none_uri`` prevents the
parser from trying to resolve ``system`` URIs. ``dtd_validation`` is off —
Word files are common; they are validated explicitly by ``doced validate``.
"""
return ET.XMLParser(
no_network=True,
load_dtd=False,
dtd_validation=False,
huge_tree=False,
recover=True,
)
def parse_string(text: str | bytes) -> ET._ElementTree:
"""Parse an OOXML part from a string/bytes into an ElementTree."""
return ET.fromstring(text, parser=get_parser())
def parse_root(text: str | bytes) -> ET._Element:
"""Parse and return the root element of an OOXML part.
Accepts ``bytes`` (preferred — preserves the XML encoding declaration) or
``str``. Word parts carry ``<?xml ... encoding="UTF-8"?>``, which requires
bytes input to :func:`lxml.etree.fromstring`.
"""
if isinstance(text, str):
text = text.encode("utf-8")
root = ET.fromstring(text, parser=get_parser())
if root is None:
raise ValueError("empty or malformed XML part")
return root
def tostring(
element: ET._Element,
*,
xml_declaration: bool = False,
pretty_print: bool = False,
) -> bytes:
"""Serialize an element to bytes, preserving namespaces.
``xml_declaration`` emits ``<?xml version="1.0" encoding="UTF-8"?>`` which
each OPC part carries. ``pretty_print`` is deliberately *off* by default —
DOCX fidelity depends on compact, byte-stable output.
"""
return ET.tostring(
element,
xml_declaration=xml_declaration,
encoding="UTF-8",
pretty_print=pretty_print,
)
def find(element: ET._Element, path: str) -> ET._Element | None:
"""Namespace-aware ``find`` using the canonical map prefixes."""
return element.find(path, namespaces=NSMAP)
def findall(element: ET._Element, path: str) -> list[ET._Element]:
"""Namespace-aware ``findall`` using the canonical map prefixes."""
return element.findall(path, namespaces=NSMAP)
def iter_all(element: ET._Element, tag: str) -> list[ET._Element]:
"""Iterate all descendants with a given *local* name, namespace-agnostic."""
local_name = local(tag)
return [e for e in element.iter() if local(e.tag) == local_name]
+933
View File
@@ -0,0 +1,933 @@
from __future__ import annotations
import base64
import json
import os
import re
import tempfile
import uuid
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Union
import regex as re_lib
from docx import Document
from docx.shared import Inches, Pt, RGBColor
from docx.enum.text import WD_ALIGN_PARAGRAPH
from PIL import Image as PILImage
from io import BytesIO
from security import SecurityConfig
def file_to_result(path: str, return_content: bool = False) -> Any:
"""
Helper: if return_content is True, read file and return {path, size, content_base64}.
Otherwise return {path, size}.
"""
size = os.path.getsize(path)
if not return_content:
return {"path": path, "size": size}
with open(path, "rb") as f:
data = f.read()
b64 = base64.b64encode(data).decode("utf-8")
return {"path": path, "size": len(data), "content_base64": b64}
@dataclass
class DocumentMetadata:
document_id: str
path: str
name: str
size: int
pages: int
class DocxToolsProvider:
def __init__(
self,
security_config: SecurityConfig,
templates_dir: str,
):
self.security_config = security_config
self.templates_dir = templates_dir
self.documents: Dict[str, Any] = {}
self._temp_base = tempfile.mkdtemp(prefix="py_docx_mcp_")
# ---- basic lifecycle ----
def create_document(self) -> str:
doc_id = str(uuid.uuid4())
path = os.path.join(self._temp_base, f"{doc_id}.docx")
doc = Document()
doc.save(path)
self.documents[doc_id] = {
"doc": doc,
"path": path,
"name": "Untitled",
}
return doc_id
def open_document(self, path: str) -> str:
if not os.path.isfile(path):
raise ValueError(f"File not found: {path}")
doc_id = str(uuid.uuid4())
doc = Document(path)
self.documents[doc_id] = {
"doc": doc,
"path": path,
"name": os.path.basename(path),
}
return doc_id
def get_doc(self, document_id: str) -> Document:
entry = self.documents.get(document_id)
if not entry:
raise ValueError(f"Document not found: {document_id}")
return entry["doc"]
def list_documents(self) -> List[Dict[str, Any]]:
out = []
for doc_id, info in self.documents.items():
out.append({
"document_id": doc_id,
"name": info["name"],
"path": info["path"],
})
return out
def close_document(self, document_id: str) -> None:
if document_id not in self.documents:
raise ValueError("Document not found")
del self.documents[document_id]
# ---- content operations ----
def add_paragraph(
self,
document_id: str,
text: str,
style: Dict[str, Any],
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
p = doc.add_paragraph(text)
run = p.runs[0] if p.runs else p.add_run()
font_family = style.get("font_family")
font_size = style.get("font_size")
bold = style.get("bold")
italic = style.get("italic")
underline = style.get("underline")
color = style.get("color")
alignment = style.get("alignment")
if font_family:
run.font.name = font_family
if font_size:
run.font.size = Pt(font_size)
if bold is not None:
run.bold = bool(bold)
if italic is not None:
run.italic = bool(italic)
if underline is not None:
run.underline = bool(underline)
if color:
try:
run.font.color.rgb = RGBColor.from_string(color)
except Exception:
pass
if alignment:
align = alignment.lower()
if align == "center":
p.alignment = WD_ALIGN_PARAGRAPH.CENTER
elif align == "right":
p.alignment = WD_ALIGN_PARAGRAPH.RIGHT
elif align == "justify":
p.alignment = WD_ALIGN_PARAGRAPH.JUSTIFY
return self._maybe_return_doc(document_id, return_content)
def add_heading(
self,
document_id: str,
text: str,
level: int,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
level = max(0, min(6, level))
doc.add_heading(text, level=level)
return self._maybe_return_doc(document_id, return_content)
def add_table(
self,
document_id: str,
rows: List[List[str]],
headers: Optional[List[str]] = None,
border_style: Optional[str] = None,
col_widths: Optional[List[int]] = None,
cell_shading: Optional[str] = None,
merges: Optional[List[Dict[str, int]]] = None,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
rows = rows or []
if headers:
rows = [headers] + rows
if not rows:
return self._maybe_return_doc(document_id, return_content)
table = doc.add_table(rows=len(rows), cols=len(rows[0]))
for ri, row in enumerate(rows):
for ci, val in enumerate(row):
table.cell(ri, ci).text = str(val or "")
if merges:
for m in merges:
r = m.get("row", 0)
c = m.get("col", 0)
row_span = m.get("row_span", 1)
col_span = m.get("col_span", 1)
if row_span > 1 or col_span > 1:
table.cell(r, c).merge(
table.cell(r + row_span - 1, c + col_span - 1)
)
return self._maybe_return_doc(document_id, return_content)
def add_section_break(
self,
document_id: str,
page_size: Optional[str] = None,
orientation: Optional[str] = None,
margins: Optional[Dict[str, float]] = None,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
doc.add_page_break()
return self._maybe_return_doc(document_id, return_content)
def add_list(
self,
document_id: str,
items: List[str],
ordered: bool,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
prefix = "1. " if ordered else "- "
for item in items:
doc.add_paragraph(f"{prefix}{item}")
return self._maybe_return_doc(document_id, return_content)
def add_list_item(
self,
document_id: str,
text: str,
level: int,
ordered: bool,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
indent = " " * level
prefix = "1. " if ordered else "- "
doc.add_paragraph(f"{indent}{prefix}{text}")
return self._maybe_return_doc(document_id, return_content)
def add_page_break(
self,
document_id: str,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
doc.add_page_break()
return self._maybe_return_doc(document_id, return_content)
def insert_toc(
self,
document_id: str,
from_level: int = 1,
to_level: int = 3,
right_align_dots: bool = True,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
doc.add_paragraph("Table of Contents (placeholder)")
return self._maybe_return_doc(document_id, return_content)
def insert_bookmark_after_heading(
self,
document_id: str,
heading_text: str,
name: str,
return_content: bool = False,
) -> Any:
# python-docx does not expose bookmarks easily; placeholder.
return self._maybe_return_doc(document_id, return_content)
def set_header(
self,
document_id: str,
text: str,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
section = doc.sections[0]
header = section.header
header.paragraphs[0].text = text
return self._maybe_return_doc(document_id, return_content)
def set_footer(
self,
document_id: str,
text: str,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
section = doc.sections[0]
footer = section.footer
footer.paragraphs[0].text = text
return self._maybe_return_doc(document_id, return_content)
def set_page_numbering(
self,
document_id: str,
location: str,
template: Optional[str] = None,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
section = doc.sections[0]
target = section.footer if location == "footer" else section.header
target.paragraphs[0].text = template or "Page {PAGE} of {PAGES}"
return self._maybe_return_doc(document_id, return_content)
def embed_page_number_fields(
self,
document_id: str,
return_content: bool = False,
) -> Any:
# python-docx cannot easily inject raw field codes; no-op placeholder.
return self._maybe_return_doc(document_id, return_content)
def add_image(
self,
document_id: str,
data_base64: str,
width: Optional[int] = None,
height: Optional[int] = None,
alt_text: Optional[str] = None,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
img_data = base64.b64decode(data_base64)
img = PILImage.open(BytesIO(img_data))
tmp_path = "/tmp/py_docx_temp.png"
img.save(tmp_path, format="PNG")
doc.add_picture(
tmp_path,
width=Inches(width / 96.0) if width else None,
height=Inches(height / 96.0) if height else None,
)
return self._maybe_return_doc(document_id, return_content)
def add_hyperlink(
self,
document_id: str,
text: str,
url: str,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
doc.add_paragraph(f"{text} ({url})")
return self._maybe_return_doc(document_id, return_content)
def find_and_replace(
self,
document_id: str,
find_text: str,
replace_text: str,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
count = 0
for para in doc.paragraphs:
if find_text in para.text:
para.text = para.text.replace(find_text, replace_text)
count += para.text.count(replace_text)
return {
"success": True,
"replacements": count,
"document": self._maybe_return_doc(document_id, return_content),
}
def find_and_replace_advanced(
self,
document_id: str,
pattern: str,
replacement: str,
case_sensitive: bool,
whole_word: bool,
use_regex: bool,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
count = 0
for para in doc.paragraphs:
original = para.text
if use_regex:
flags = 0 if case_sensitive else re_lib.IGNORECASE
pat = pattern
else:
if whole_word:
pat = r"\b" + re_lib.escape(pattern) + r"\b"
else:
pat = re_lib.escape(pattern)
flags = 0 if case_sensitive else re_lib.IGNORECASE
new_text, n = re_lib.subn(pat, replacement, original, flags=flags)
if new_text != original:
para.text = new_text
count += n
return {
"success": True,
"replacements": count,
"document": self._maybe_return_doc(document_id, return_content),
}
def apply_paragraph_format(
self,
document_id: str,
contains: Optional[str],
format: Dict[str, Any],
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
updated = 0
for para in doc.paragraphs:
if contains and (contains not in para.text):
continue
for run in para.runs or []:
if "font_family" in format:
run.font.name = format["font_family"]
if "font_size" in format:
run.font.size = Pt(int(format["font_size"]))
if "bold" in format:
run.bold = bool(format["bold"])
if "italic" in format:
run.italic = bool(format["italic"])
if "underline" in format:
run.underline = bool(format["underline"])
if "color" in format:
try:
run.font.color.rgb = RGBColor.from_string(format["color"])
except Exception:
pass
updated += 1
return {
"success": True,
"paragraphs_updated": updated,
"document": self._maybe_return_doc(document_id, return_content),
}
def extract_text(self, document_id: str) -> str:
doc = self.get_doc(document_id)
return "\n".join(p.text for p in doc.paragraphs)
def get_tables(self, document_id: str) -> List[Dict[str, Any]]:
doc = self.get_doc(document_id)
out = []
for idx, table in enumerate(doc.tables):
rows_data = []
for row in table.rows:
cells = [cell.text for cell in row.cells]
rows_data.append(cells)
out.append({
"index": idx,
"rows": len(table.rows),
"cols": len(table.columns),
"data": rows_data,
})
return out
def list_images(self, document_id: str) -> List[Dict[str, Any]]:
return []
def list_hyperlinks(self, document_id: str) -> List[Dict[str, Any]]:
doc = self.get_doc(document_id)
links = []
for p in doc.paragraphs:
for m in re.finditer(r"\((https?://\S+)\)", p.text):
links.append({"text": p.text.strip(), "url": m.group(1)})
return links
def get_fields_summary(self, document_id: str) -> Dict[str, Any]:
return {"note": "Fields summary not fully implemented in Python version"}
def strip_personal_info(self, document_id: str) -> None:
doc = self.get_doc(document_id)
core = doc.core_properties
core.author = ""
core.last_modified_by = ""
core.revision_number = 1
def get_metadata(self, document_id: str) -> DocumentMetadata:
info = self.documents[document_id]
path = info["path"]
size = os.path.getsize(path)
doc = info["doc"]
pages = max(1, len(doc.paragraphs) // 40)
return DocumentMetadata(
document_id=document_id,
path=path,
name=info["name"],
size=size,
pages=pages,
)
def save_document(
self,
document_id: str,
output_path: str,
return_content: bool = True,
) -> Any:
info = self.documents[document_id]
os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
info["doc"].save(output_path)
return file_to_result(output_path, return_content=return_content)
# ---- conversion (best-effort, external tools optional) ----
def convert_to_pdf(
self,
document_id: str,
output_path: str,
prefer_external: bool = False,
return_content: bool = True,
) -> Any:
meta = self.get_metadata(document_id)
# If external tools are allowed, you can call LibreOffice here;
# for now, we indicate requirement.
if prefer_external:
raise NotImplementedError(
"External PDF conversion not yet wired; "
"configure LibreOffice/unoconv integration."
)
raise NotImplementedError(
"PDF conversion not yet implemented in pure Python version."
)
def export_pdf_with_field_refresh(
self,
document_id: str,
output_path: str,
prefer_external: bool = True,
return_content: bool = True,
) -> Any:
self.embed_page_number_fields(document_id)
return self.convert_to_pdf(document_id, output_path, prefer_external, return_content=return_content)
def convert_to_images(
self,
document_id: str,
output_dir: str,
format: str = "png",
dpi: int = 150,
return_content: bool = True,
) -> Any:
raise NotImplementedError(
"Image conversion not yet implemented in pure Python version."
)
def convert_to_images_with_preference(
self,
document_id: str,
output_dir: str,
format: str = "png",
dpi: int = 150,
prefer_external: bool = True,
return_content: bool = True,
) -> Any:
return self.convert_to_images(document_id, output_dir, format, dpi, return_content=return_content)
# ---- advanced docx operations ----
def merge_documents(
self,
document_ids: List[str],
output_path: str,
return_content: bool = True,
) -> Any:
merged = Document()
for did in document_ids:
doc = self.get_doc(did)
for elem in doc.element.body:
merged.element.body.append(elem)
os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
merged.save(output_path)
return file_to_result(output_path, return_content=return_content)
def split_document(
self,
document_id: str,
output_dir: str,
return_content: bool = True,
) -> Any:
# Very naive: split by page breaks.
doc = self.get_doc(document_id)
os.makedirs(output_dir, exist_ok=True)
parts: List[Document] = []
current = Document()
for elem in doc.element.body:
tag = elem.tag
if "lastRenderedPageBreak" in tag or "pageBreakBefore" in tag:
parts.append(current)
current = Document()
else:
current.element.body.append(elem)
if len(current.element.body) > 0:
parts.append(current)
results = []
for i, pdoc in enumerate(parts):
path = os.path.join(output_dir, f"part_{i}.docx")
pdoc.save(path)
results.append(file_to_result(path, return_content=return_content))
return {"parts": results}
def get_document_structure(self, document_id: str) -> Dict[str, Any]:
doc = self.get_doc(document_id)
headings = []
for p in doc.paragraphs:
if p.style.name.startswith("Heading"):
headings.append({
"text": p.text,
"style": p.style.name,
})
return {
"headings": headings,
"paragraph_count": len(doc.paragraphs),
"table_count": len(doc.tables),
}
def get_outline(self, document_id: str) -> List[Dict[str, Any]]:
return self.get_document_structure(document_id).get("headings", [])
def get_ranges(self, document_id: str, selector: str) -> List[Dict[str, Any]]:
# Minimal: "heading:'Text'" or "paragraph[i]"
doc = self.get_doc(document_id)
ranges = []
if selector.startswith("heading:"):
target = selector[len("heading:"):].strip().strip("'\"")
for i, p in enumerate(doc.paragraphs):
if p.style.name.startswith("Heading") and target.lower() in p.text.lower():
ranges.append({"type": "paragraph", "index": i})
elif selector.startswith("paragraph["):
m = re.match(r"paragraph\[(\d+)\]", selector)
if m:
idx = int(m.group(1))
ranges.append({"type": "paragraph", "index": idx})
return ranges
def replace_range_text(
self,
document_id: str,
range_id: Dict[str, Any],
text: str,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
if range_id.get("type") == "paragraph":
idx = range_id.get("index")
if 0 <= idx < len(doc.paragraphs):
doc.paragraphs[idx].text = text
return self._maybe_return_doc(document_id, return_content)
def set_table_cell_text(
self,
document_id: str,
table_index: int,
row: int,
col: int,
text: str,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
table = doc.tables[table_index]
table.cell(row, col).text = text
return self._maybe_return_doc(document_id, return_content)
def get_document_properties(self, document_id: str) -> Dict[str, Any]:
doc = self.get_doc(document_id)
core = doc.core_properties
return {
"title": core.title,
"subject": core.subject,
"author": core.author,
"last_modified_by": core.last_modified_by,
"created": str(core.created),
"modified": str(core.modified),
}
def set_document_properties(
self,
document_id: str,
title: Optional[str],
subject: Optional[str],
author: Optional[str],
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
core = doc.core_properties
if title is not None:
core.title = title
if subject is not None:
core.subject = subject
if author is not None:
core.author = author
return self._maybe_return_doc(document_id, return_content)
def insert_after_heading(
self,
document_id: str,
heading_text: str,
text: str,
return_content: bool = False,
) -> Any:
doc = self.get_doc(document_id)
for p in doc.paragraphs:
if p.style.name.startswith("Heading") and heading_text.lower() in p.text.lower():
doc.add_paragraph(text)
return self._maybe_return_doc(document_id, return_content)
return {"success": False, "reason": "Heading not found"}
def sanitize_external_links(self, document_id: str) -> None:
# naive: remove URLs from text
doc = self.get_doc(document_id)
for p in doc.paragraphs:
p.text = re.sub(r"(https?://\S+)", "", p.text)
def redact_text(
self,
document_id: str,
pattern: str,
use_regex: bool = False,
whole_word: bool = False,
case_sensitive: bool = False,
return_content: bool = False,
) -> Any:
result = self.find_and_replace_advanced(
document_id,
pattern=pattern,
replacement="█",
case_sensitive=case_sensitive,
whole_word=whole_word,
use_regex=use_regex,
return_content=return_content,
)
return result
def analyze_formatting(self, document_id: str) -> Dict[str, Any]:
doc = self.get_doc(document_id)
styles = set()
fonts = set()
for p in doc.paragraphs:
styles.add(p.style.name)
for run in p.runs or []:
if run.font.name:
fonts.add(run.font.name)
return {
"styles_used": list(styles),
"fonts_detected": list(fonts),
"has_tables": len(doc.tables) > 0,
"has_images": False,
"has_hyperlinks": any(
"http" in p.text.lower() for p in doc.paragraphs
),
"page_count": max(1, len(doc.paragraphs) // 40),
"section_count": len(doc.sections),
}
def get_word_count(self, document_id: str) -> Dict[str, Any]:
text = self.extract_text(document_id)
words = text.split()
chars = len(text)
chars_no_spaces = len(text.replace(" ", ""))
paragraphs = len([l for l in text.splitlines() if l.strip()])
sentences = len(re.findall(r"[.!?]+", text))
return {
"words": len(words),
"characters": chars,
"characters_no_spaces": chars_no_spaces,
"paragraphs": paragraphs,
"sentences": sentences,
"pages": max(1, len(words) // 250),
"reading_time_minutes": max(1, len(words) // 200),
}
def search_text(
self,
document_id: str,
search_term: str,
case_sensitive: bool = False,
whole_word: bool = False,
) -> Dict[str, Any]:
text = self.extract_text(document_id)
if not case_sensitive:
text_lower = text.lower()
term_lower = search_term.lower()
else:
text_lower = text
term_lower = search_term
if whole_word:
pattern = r"\b" + re_lib.escape(term_lower) + r"\b"
else:
pattern = re_lib.escape(term_lower)
matches = []
for m in re_lib.finditer(pattern, text_lower):
start = max(0, m.start() - 50)
end = min(len(text), m.end() + 50)
line = text[: m.start()].count("\n") + 1
matches.append({
"position": m.start(),
"context": text[start:end],
"line": line,
})
return {
"matches": matches,
"total_matches": len(matches),
}
def export_to_markdown(
self,
document_id: str,
output_path: str,
return_content: bool = True,
) -> Any:
text = self.extract_text(document_id)
md_lines = []
for line in text.splitlines():
t = line.strip()
if not t:
md_lines.append("")
continue
if len(t) < 100 and any(c.isupper() for c in t):
if all(c.isupper() or c.isspace() for c in t):
md_lines.append(f"# {t}")
else:
md_lines.append(f"## {t}")
else:
md_lines.append(t)
md = "\n\n".join(md_lines)
os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
with open(output_path, "w", encoding="utf-8") as f:
f.write(md)
return file_to_result(output_path, return_content=return_content)
def export_to_html(
self,
document_id: str,
output_path: str,
return_content: bool = True,
) -> Any:
text = self.extract_text(document_id)
html_parts = ['<html><head><meta charset="utf-8"></head><body>\n']
for line in text.splitlines():
t = line.strip()
if not t:
continue
if len(t) < 100 and any(c.isupper() for c in t):
if all(c.isupper() or c.isspace() for c in t):
html_parts.append(f"<h1>{t}</h1>")
else:
html_parts.append(f"<h2>{t}</h2>")
elif t.startswith("- ") or t.startswith("* "):
html_parts.append(f"<ul><li>{t[2:]}</li></ul>")
else:
html_parts.append(f"<p>{t}</p>")
html_parts.append("</body></html>\n")
html = "\n".join(html_parts)
os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
with open(output_path, "w", encoding="utf-8") as f:
f.write(html)
return file_to_result(output_path, return_content=return_content)
# ---- security and storage info ----
def get_security_info(self) -> Dict[str, Any]:
return {
"readonly_mode": self.security_config.readonly_mode,
"sandbox_mode": self.security_config.sandbox_mode,
"allow_external_tools": self.security_config.allow_external_tools,
"allow_network": self.security_config.allow_network,
"max_document_size": self.security_config.max_document_size,
"max_open_documents": self.security_config.max_open_documents,
}
def get_storage_info(self) -> Dict[str, Any]:
total = 0
for info in self.documents.values():
try:
total += os.path.getsize(info["path"])
except OSError:
pass
return {
"temp_base": self._temp_base,
"open_documents": len(self.documents),
"total_size_bytes": total,
}
# ---- templates ----
def open_template(self, name: str, templates_dir: str) -> str:
path = os.path.join(templates_dir, name)
if not os.path.isfile(path):
raise ValueError(f"Template not found: {name}")
return self.open_document(path)
def generate_from_template(
self,
template_name: str,
output_path: str,
fields: Dict[str, str],
return_content: bool = True,
) -> Any:
template_path = os.path.join(self.templates_dir, template_name)
if not os.path.isfile(template_path):
raise ValueError(f"Template not found: {template_name}")
doc_id = self.open_document(template_path)
for key, value in fields.items():
placeholder = "{{" + key + "}}"
self.find_and_replace_advanced(
doc_id,
pattern=placeholder,
replacement=str(value),
case_sensitive=False,
whole_word=True,
use_regex=False,
return_content=False,
)
self.save_document(doc_id, output_path, return_content=False)
self.close_document(doc_id)
return file_to_result(output_path, return_content=return_content)
# ---- internal helper ----
def _maybe_return_doc(
self,
document_id: str,
return_content: bool,
) -> Any:
"""
If return_content is True, save the current document in-memory state
to its path and return base64 content.
"""
if not return_content:
return {"success": True, "document_id": document_id}
info = self.documents[document_id]
info["doc"].save(info["path"])
return file_to_result(info["path"], return_content=True)
+96
View File
@@ -0,0 +1,96 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Set
@dataclass
class SecurityConfig:
readonly_mode: bool
sandbox_mode: bool
allow_external_tools: bool
allow_network: bool
max_document_size: int
max_open_documents: int
def get_summary(self) -> str:
parts = []
if self.readonly_mode:
parts.append("readonly")
if self.sandbox_mode:
parts.append("sandbox")
if self.allow_external_tools:
parts.append("external-tools")
if self.allow_network:
parts.append("network")
return ", ".join(parts) or "default"
# Tools allowed in readonly mode
READONLY_COMMANDS: Set[str] = {
"list_documents",
"open_document",
"extract_text",
"get_metadata",
"get_document_structure",
"get_outline",
"get_ranges",
"get_tables",
"list_images",
"list_hyperlinks",
"get_fields_summary",
"get_document_properties",
"get_word_count",
"search_text",
"analyze_formatting",
"get_security_info",
"get_storage_info",
"list_templates",
}
# Tools that modify documents
WRITE_COMMANDS: Set[str] = {
"create_document",
"add_paragraph",
"add_heading",
"add_table",
"add_section_break",
"add_list",
"add_list_item",
"add_page_break",
"insert_toc",
"insert_bookmark_after_heading",
"set_header",
"set_footer",
"set_page_numbering",
"embed_page_number_fields",
"add_image",
"add_hyperlink",
"find_and_replace",
"find_and_replace_advanced",
"apply_paragraph_format",
"save_document",
"close_document",
"convert_to_pdf",
"export_pdf_with_field_refresh",
"convert_to_images",
"convert_to_images_with_preference",
"merge_documents",
"split_document",
"replace_range_text",
"set_table_cell_text",
"set_document_properties",
"insert_after_heading",
"sanitize_external_links",
"redact_text",
"strip_personal_info",
"export_to_markdown",
"export_to_html",
"open_template",
"generate_from_template",
}
def is_command_allowed(name: str, config: SecurityConfig) -> bool:
if config.readonly_mode:
return name in READONLY_COMMANDS
return True
+916
View File
@@ -0,0 +1,916 @@
from __future__ import annotations
import json
import logging
import os
from typing import Any, Dict, Optional
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from docx_tools import DocxToolsProvider
from security import SecurityConfig, is_command_allowed
from templates import list_templates
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("py-docx-mcp")
TEMPLATES_DIR = os.getenv("DOCX_MCP_TEMPLATES_DIR", "/templates")
def build_tools_list(provider: DocxToolsProvider, security_config: SecurityConfig) -> Dict[str, Any]:
"""
Build the tools list returned by tools/list,
matching the shape expected by OpenWebUI.
"""
def allowed(name: str) -> bool:
return is_command_allowed(name, security_config)
tools = []
def add(name: str, desc: str, required: list[str],
extra_props: Optional[Dict[str, Any]] = None):
if not allowed(name):
return
props = dict(extra_props or {})
for r in required:
if r not in props:
props[r] = {"type": "string"}
tools.append({
"name": name,
"description": desc,
"inputSchema": {
"type": "object",
"properties": props,
"required": required,
},
})
add("create_document",
"Create a new empty DOCX document",
[])
add("open_document",
"Open an existing DOCX document",
["path"],
{"path": {"type": "string", "description": "Path to the DOCX file"}})
add("add_paragraph",
"Add a paragraph with optional styling to the document",
["document_id", "text"],
{
"document_id": {"type": "string"},
"text": {"type": "string"},
"style": {
"type": "object",
"description": "Paragraph style options (font_family, font_size, bold, italic, underline, color, alignment)"
},
"return_content": {"type": "boolean", "description": "Return document content as base64", "default": False},
})
add("add_heading",
"Add a heading to the document",
["document_id", "text", "level"],
{
"document_id": {"type": "string"},
"text": {"type": "string"},
"level": {"type": "integer", "description": "Heading level (1-6)"},
"return_content": {"type": "boolean", "default": False},
})
add("add_table",
"Add a table to the document",
["document_id", "rows"],
{
"document_id": {"type": "string"},
"rows": {"type": "array", "items": {"type": "array", "items": {"type": "string"}}},
"headers": {"type": "array", "items": {"type": "string"}},
"border_style": {"type": "string"},
"col_widths": {"type": "array", "items": {"type": "integer"}},
"cell_shading": {"type": "string"},
"merges": {"type": "array", "items": {"type": "object"}},
"return_content": {"type": "boolean", "default": False},
})
add("add_section_break",
"Insert a section break with optional page setup",
["document_id"],
{
"document_id": {"type": "string"},
"page_size": {"type": "string"},
"orientation": {"type": "string"},
"margins": {"type": "object"},
"return_content": {"type": "boolean", "default": False},
})
add("add_list",
"Add a bulleted or numbered list to the document",
["document_id", "items"],
{
"document_id": {"type": "string"},
"items": {"type": "array", "items": {"type": "string"}},
"ordered": {"type": "boolean", "default": False},
"return_content": {"type": "boolean", "default": False},
})
add("add_list_item",
"Add a single list item with a specific level",
["document_id", "text"],
{
"document_id": {"type": "string"},
"text": {"type": "string"},
"level": {"type": "integer", "default": 0},
"ordered": {"type": "boolean", "default": False},
"return_content": {"type": "boolean", "default": False},
})
add("add_page_break",
"Add a page break to the document",
["document_id"],
{"return_content": {"type": "boolean", "default": False}})
add("insert_toc",
"Insert a Table of Contents placeholder",
["document_id"],
{
"from_level": {"type": "integer", "default": 1},
"to_level": {"type": "integer", "default": 3},
"right_align_dots": {"type": "boolean", "default": True},
"return_content": {"type": "boolean", "default": False},
})
add("insert_bookmark_after_heading",
"Insert a bookmark immediately after the first matching heading",
["document_id", "heading_text", "name"],
{"return_content": {"type": "boolean", "default": False}})
add("set_header",
"Set the document header",
["document_id", "text"],
{"return_content": {"type": "boolean", "default": False}})
add("set_footer",
"Set the document footer",
["document_id", "text"],
{"return_content": {"type": "boolean", "default": False}})
add("set_page_numbering",
"Set a simple page numbering text in header or footer",
["document_id"],
{
"location": {"type": "string", "default": "footer"},
"template": {"type": "string"},
"return_content": {"type": "boolean", "default": False},
})
add("embed_page_number_fields",
"Replace placeholder 'Page {PAGE} of {PAGES}' with Word field codes (best-effort)",
["document_id"],
{"return_content": {"type": "boolean", "default": False}})
add("add_image",
"Insert an image into the document",
["document_id", "data_base64"],
{
"data_base64": {"type": "string"},
"width": {"type": "integer"},
"height": {"type": "integer"},
"alt_text": {"type": "string"},
"return_content": {"type": "boolean", "default": False},
})
add("add_hyperlink",
"Insert a hyperlink into the document",
["document_id", "text", "url"],
{"return_content": {"type": "boolean", "default": False}})
add("find_and_replace",
"Find and replace text in the document",
["document_id", "find_text", "replace_text"],
{"return_content": {"type": "boolean", "default": False}})
add("find_and_replace_advanced",
"Find/replace with regex, case, whole-word, preserving runs",
["document_id", "pattern", "replacement"],
{
"case_sensitive": {"type": "boolean", "default": False},
"whole_word": {"type": "boolean", "default": False},
"use_regex": {"type": "boolean", "default": False},
"return_content": {"type": "boolean", "default": False},
})
add("apply_paragraph_format",
"Apply paragraph formatting to paragraphs matching a simple selector",
["document_id"],
{
"contains": {"type": "string"},
"format": {"type": "object"},
"return_content": {"type": "boolean", "default": False},
})
add("extract_text",
"Extract all text content from the document",
["document_id"])
add("get_tables",
"List tables with dimensions, merges, and cell content",
["document_id"])
add("list_images",
"List images with width/height and alt text",
["document_id"])
add("list_hyperlinks",
"List hyperlinks in the document",
["document_id"])
add("get_fields_summary",
"Summarize Word fields (PAGE, NUMPAGES, TOC) in document and headers/footers",
["document_id"])
add("strip_personal_info",
"Remove personal info from metadata and core.xml (best-effort)",
["document_id"])
add("get_metadata",
"Get document metadata",
["document_id"])
add("save_document",
"Save the document to a specific path and return its content",
["document_id", "output_path"],
{"return_content": {"type": "boolean", "default": True}})
add("close_document",
"Close the document and free resources",
["document_id"])
add("list_documents",
"List all open documents",
[])
add("convert_to_pdf",
"Convert a DOCX document to PDF and return the file",
["document_id", "output_path"],
{
"prefer_external": {"type": "boolean", "default": False},
"return_content": {"type": "boolean", "default": True},
})
add("export_pdf_with_field_refresh",
"Embed page fields then export to PDF (hi-fidelity when available)",
["document_id", "output_path"],
{
"prefer_external": {"type": "boolean", "default": True},
"return_content": {"type": "boolean", "default": True},
})
add("convert_to_images",
"Convert a DOCX document to images (one per page) and return them",
["document_id", "output_dir"],
{
"format": {"type": "string", "default": "png"},
"dpi": {"type": "integer", "default": 150},
"return_content": {"type": "boolean", "default": True},
})
add("convert_to_images_with_preference",
"Convert DOCX to images, preferring external hi-fidelity path",
["document_id", "output_dir"],
{
"format": {"type": "string", "default": "png"},
"dpi": {"type": "integer", "default": 150},
"prefer_external": {"type": "boolean", "default": True},
"return_content": {"type": "boolean", "default": True},
})
add("merge_documents",
"Merge multiple DOCX documents into one and return the result",
["document_ids", "output_path"],
{
"document_ids": {"type": "array", "items": {"type": "string"}},
"return_content": {"type": "boolean", "default": True},
})
add("split_document",
"Split a document at page breaks and return parts",
["document_id", "output_dir"],
{"return_content": {"type": "boolean", "default": True}})
add("get_document_structure",
"Get the structural overview of the document (headings, sections, etc.)",
["document_id"])
add("get_outline",
"Return heading outline with range_ids",
["document_id"])
add("get_ranges",
"Resolve a selector to range_ids",
["document_id", "selector"])
add("replace_range_text",
"Replace text in a paragraph/heading by range_id",
["document_id", "range_id", "text"],
{"return_content": {"type": "boolean", "default": False}})
add("set_table_cell_text",
"Set text in a table cell by indices",
["document_id", "table_index", "row", "col", "text"],
{
"table_index": {"type": "integer"},
"row": {"type": "integer"},
"col": {"type": "integer"},
"return_content": {"type": "boolean", "default": False},
})
add("get_document_properties",
"Get document properties (title, subject, author, timestamps)",
["document_id"])
add("set_document_properties",
"Set document properties (title, subject, author)",
["document_id"],
{
"title": {"type": "string"},
"subject": {"type": "string"},
"author": {"type": "string"},
"return_content": {"type": "boolean", "default": False},
})
add("insert_after_heading",
"Insert a paragraph after the first heading that matches text",
["document_id", "heading_text", "text"],
{"return_content": {"type": "boolean", "default": False}})
add("sanitize_external_links",
"Remove external hyperlinks (http/https)",
["document_id"])
add("redact_text",
"Redact text using regex/whole-word with █ character",
["document_id", "pattern"],
{
"use_regex": {"type": "boolean", "default": False},
"whole_word": {"type": "boolean", "default": False},
"case_sensitive": {"type": "boolean", "default": False},
"return_content": {"type": "boolean", "default": False},
})
add("analyze_formatting",
"Analyze the formatting used throughout the document",
["document_id"])
add("get_word_count",
"Get detailed word count statistics for the document",
["document_id"])
add("search_text",
"Search for text patterns in the document",
["document_id", "search_term"],
{
"case_sensitive": {"type": "boolean", "default": False},
"whole_word": {"type": "boolean", "default": False},
})
add("export_to_markdown",
"Export document content to Markdown format and return the file",
["document_id", "output_path"],
{"return_content": {"type": "boolean", "default": True}})
add("export_to_html",
"Export document content to HTML format and return the file",
["document_id", "output_path"],
{"return_content": {"type": "boolean", "default": True}})
add("get_security_info",
"Get information about current security settings and restrictions",
[])
add("get_storage_info",
"Get information about temporary storage usage",
[])
add("list_templates",
"List available document templates from the templates directory",
[])
add("open_template",
"Open a template document by name from the templates directory",
["name"])
add("generate_from_template",
"Generate a new document from a template and return the file",
["template_name", "output_path", "fields"],
{
"fields": {"type": "object"},
"return_content": {"type": "boolean", "default": True},
})
return {"tools": tools}
def make_app() -> FastAPI:
app = FastAPI(title="py-docx-mcp")
readonly_mode = os.getenv("DOCX_MCP_READONLY", "false").lower() in ("true", "1")
sandbox_mode = os.getenv("DOCX_MCP_SANDBOX", "true").lower() in ("true", "1")
allow_external_tools = os.getenv("DOCX_MCP_ALLOW_EXTERNAL_TOOLS", "false").lower() in ("true", "1")
allow_network = os.getenv("DOCX_MCP_ALLOW_NETWORK", "false").lower() in ("true", "1")
max_document_size = int(os.getenv("DOCX_MCP_MAX_SIZE", "104857600"))
max_open_documents = int(os.getenv("DOCX_MCP_MAX_DOCS", "30"))
api_key = os.getenv("DOCX_MCP_API_KEY", "").strip()
security_config = SecurityConfig(
readonly_mode=readonly_mode,
sandbox_mode=sandbox_mode,
allow_external_tools=allow_external_tools,
allow_network=allow_network,
max_document_size=max_document_size,
max_open_documents=max_open_documents,
)
provider = DocxToolsProvider(
security_config=security_config,
templates_dir=TEMPLATES_DIR,
)
tools_list = build_tools_list(provider, security_config)
def get_bearer_token(request: Request) -> Optional[str]:
auth = (request.headers.get("Authorization") or "").strip()
if auth.startswith("Bearer "):
return auth[len("Bearer "):].strip()
return None
def require_auth(request: Request) -> bool:
if not api_key:
return True
token = get_bearer_token(request)
if not token or token != api_key:
return False
return True
@app.get("/")
async def root(request: Request):
if not require_auth(request):
return JSONResponse(status_code=401, content={"error": "Missing or invalid API key"})
return {
"service": "py-docx-mcp",
"transport": "streamable-http",
"docs": "Use POST / with MCP JSON-RPC (initialize, tools/list, tools/call).",
}
@app.post("/")
async def mcp_endpoint(request: Request):
if not require_auth(request):
return JSONResponse(status_code=401, content={"error": "Missing or invalid API key"})
body = await request.json()
method = body.get("method")
params = body.get("params") or {}
req_id = body.get("id")
# MCP: initialize
if method == "initialize":
return {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"protocolVersion": "2025-11-25",
"capabilities": {"tools": {}},
"serverInfo": {
"name": "py-docx-mcp",
"version": "0.1.0",
},
},
}
# MCP: tools/list
if method == "tools/list":
return {
"jsonrpc": "2.0",
"id": req_id,
"result": tools_list,
}
# MCP: tools/call
if method == "tools/call":
tool_name = params.get("name")
tool_args = params.get("arguments") or {}
# Security check: only allowed commands
if not is_command_allowed(tool_name, security_config):
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {
"code": -32000,
"message": f"Command '{tool_name}' not allowed by security policy",
},
}
try:
result = call_tool_impl(tool_name, tool_args, provider)
return {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"content": [
{
"type": "text",
"text": (result if isinstance(result, str) else json.dumps(result, ensure_ascii=False)),
}
]
},
}
except Exception as e:
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {
"code": -32000,
"message": str(e),
},
}
# Unknown method
return JSONResponse(
status_code=400,
content={"error": "Unknown method: " + str(method)},
)
return app
def call_tool_impl(name: str, args: Dict[str, Any], provider: DocxToolsProvider) -> Any:
# Delegate to provider methods, matching names.
# This central dispatcher keeps tool signatures in one place.
if name == "create_document":
return provider.create_document()
if name == "open_document":
return provider.open_document(args["path"])
if name == "add_paragraph":
return provider.add_paragraph(
args["document_id"],
args["text"],
args.get("style") or {},
return_content=args.get("return_content", False),
)
if name == "add_heading":
return provider.add_heading(
args["document_id"],
args["text"],
int(args["level"]),
return_content=args.get("return_content", False),
)
if name == "add_table":
return provider.add_table(
args["document_id"],
args["rows"],
headers=args.get("headers"),
border_style=args.get("border_style"),
col_widths=args.get("col_widths"),
cell_shading=args.get("cell_shading"),
merges=args.get("merges"),
return_content=args.get("return_content", False),
)
if name == "add_section_break":
return provider.add_section_break(
args["document_id"],
page_size=args.get("page_size"),
orientation=args.get("orientation"),
margins=args.get("margins"),
return_content=args.get("return_content", False),
)
if name == "add_list":
return provider.add_list(
args["document_id"],
args["items"],
ordered=bool(args.get("ordered", False)),
return_content=args.get("return_content", False),
)
if name == "add_list_item":
return provider.add_list_item(
args["document_id"],
args["text"],
level=int(args.get("level", 0)),
ordered=bool(args.get("ordered", False)),
return_content=args.get("return_content", False),
)
if name == "add_page_break":
return provider.add_page_break(
args["document_id"],
return_content=args.get("return_content", False),
)
if name == "insert_toc":
return provider.insert_toc(
args["document_id"],
from_level=int(args.get("from_level", 1)),
to_level=int(args.get("to_level", 3)),
right_align_dots=bool(args.get("right_align_dots", True)),
return_content=args.get("return_content", False),
)
if name == "insert_bookmark_after_heading":
return provider.insert_bookmark_after_heading(
args["document_id"],
args["heading_text"],
args["name"],
return_content=args.get("return_content", False),
)
if name == "set_header":
return provider.set_header(
args["document_id"],
args["text"],
return_content=args.get("return_content", False),
)
if name == "set_footer":
return provider.set_footer(
args["document_id"],
args["text"],
return_content=args.get("return_content", False),
)
if name == "set_page_numbering":
return provider.set_page_numbering(
args["document_id"],
location=args.get("location", "footer"),
template=args.get("template"),
return_content=args.get("return_content", False),
)
if name == "embed_page_number_fields":
return provider.embed_page_number_fields(
args["document_id"],
return_content=args.get("return_content", False),
)
if name == "add_image":
return provider.add_image(
args["document_id"],
args["data_base64"],
width=args.get("width"),
height=args.get("height"),
alt_text=args.get("alt_text"),
return_content=args.get("return_content", False),
)
if name == "add_hyperlink":
return provider.add_hyperlink(
args["document_id"],
args["text"],
args["url"],
return_content=args.get("return_content", False),
)
if name == "find_and_replace":
return provider.find_and_replace(
args["document_id"],
args["find_text"],
args["replace_text"],
return_content=args.get("return_content", False),
)
if name == "find_and_replace_advanced":
return provider.find_and_replace_advanced(
args["document_id"],
args["pattern"],
args["replacement"],
case_sensitive=bool(args.get("case_sensitive", False)),
whole_word=bool(args.get("whole_word", False)),
use_regex=bool(args.get("use_regex", False)),
return_content=args.get("return_content", False),
)
if name == "apply_paragraph_format":
return provider.apply_paragraph_format(
args["document_id"],
contains=args.get("contains"),
format=args.get("format") or {},
return_content=args.get("return_content", False),
)
if name == "extract_text":
return provider.extract_text(args["document_id"])
if name == "get_tables":
return provider.get_tables(args["document_id"])
if name == "list_images":
return provider.list_images(args["document_id"])
if name == "list_hyperlinks":
return provider.list_hyperlinks(args["document_id"])
if name == "get_fields_summary":
return provider.get_fields_summary(args["document_id"])
if name == "strip_personal_info":
return provider.strip_personal_info(args["document_id"])
if name == "get_metadata":
return provider.get_metadata(args["document_id"])
if name == "save_document":
return provider.save_document(
args["document_id"],
args["output_path"],
return_content=args.get("return_content", True),
)
if name == "close_document":
return provider.close_document(args["document_id"])
if name == "list_documents":
return provider.list_documents()
if name == "convert_to_pdf":
return provider.convert_to_pdf(
args["document_id"],
args["output_path"],
prefer_external=bool(args.get("prefer_external", False)),
return_content=args.get("return_content", True),
)
if name == "export_pdf_with_field_refresh":
return provider.export_pdf_with_field_refresh(
args["document_id"],
args["output_path"],
prefer_external=bool(args.get("prefer_external", True)),
return_content=args.get("return_content", True),
)
if name == "convert_to_images":
return provider.convert_to_images(
args["document_id"],
args["output_dir"],
format=args.get("format", "png"),
dpi=int(args.get("dpi", 150)),
return_content=args.get("return_content", True),
)
if name == "convert_to_images_with_preference":
return provider.convert_to_images_with_preference(
args["document_id"],
args["output_dir"],
format=args.get("format", "png"),
dpi=int(args.get("dpi", 150)),
prefer_external=bool(args.get("prefer_external", True)),
return_content=args.get("return_content", True),
)
if name == "merge_documents":
return provider.merge_documents(
args["document_ids"],
args["output_path"],
return_content=args.get("return_content", True),
)
if name == "split_document":
return provider.split_document(
args["document_id"],
args["output_dir"],
return_content=args.get("return_content", True),
)
if name == "get_document_structure":
return provider.get_document_structure(args["document_id"])
if name == "get_outline":
return provider.get_outline(args["document_id"])
if name == "get_ranges":
return provider.get_ranges(args["document_id"], args["selector"])
if name == "replace_range_text":
return provider.replace_range_text(
args["document_id"],
args["range_id"],
args["text"],
return_content=args.get("return_content", False),
)
if name == "set_table_cell_text":
return provider.set_table_cell_text(
args["document_id"],
int(args["table_index"]),
int(args["row"]),
int(args["col"]),
args["text"],
return_content=args.get("return_content", False),
)
if name == "get_document_properties":
return provider.get_document_properties(args["document_id"])
if name == "set_document_properties":
return provider.set_document_properties(
args["document_id"],
title=args.get("title"),
subject=args.get("subject"),
author=args.get("author"),
return_content=args.get("return_content", False),
)
if name == "insert_after_heading":
return provider.insert_after_heading(
args["document_id"],
args["heading_text"],
args["text"],
return_content=args.get("return_content", False),
)
if name == "sanitize_external_links":
return provider.sanitize_external_links(args["document_id"])
if name == "redact_text":
return provider.redact_text(
args["document_id"],
args["pattern"],
use_regex=bool(args.get("use_regex", False)),
whole_word=bool(args.get("whole_word", False)),
case_sensitive=bool(args.get("case_sensitive", False)),
return_content=args.get("return_content", False),
)
if name == "analyze_formatting":
return provider.analyze_formatting(args["document_id"])
if name == "get_word_count":
return provider.get_word_count(args["document_id"])
if name == "search_text":
return provider.search_text(
args["document_id"],
args["search_term"],
case_sensitive=bool(args.get("case_sensitive", False)),
whole_word=bool(args.get("whole_word", False)),
)
if name == "export_to_markdown":
return provider.export_to_markdown(
args["document_id"],
args["output_path"],
return_content=args.get("return_content", True),
)
if name == "export_to_html":
return provider.export_to_html(
args["document_id"],
args["output_path"],
return_content=args.get("return_content", True),
)
if name == "get_security_info":
return provider.get_security_info()
if name == "get_storage_info":
return provider.get_storage_info()
if name == "list_templates":
return list_templates(TEMPLATES_DIR)
if name == "open_template":
return provider.open_template(args["name"], TEMPLATES_DIR)
if name == "generate_from_template":
return provider.generate_from_template(
args["template_name"],
args["output_path"],
args.get("fields") or {},
return_content=args.get("return_content", True),
)
raise ValueError(f"Unknown tool: {name}")
def main():
app = make_app()
host = os.getenv("DOCX_MCP_HTTP_HOST", "0.0.0.0")
port = int(os.getenv("DOCX_MCP_HTTP_PORT", "3000"))
import uvicorn
uvicorn.run(app, host=host, port=port, log_level="info")
if __name__ == "__main__":
main()
+22
View File
@@ -0,0 +1,22 @@
from __future__ import annotations
import os
from typing import List
def list_templates(templates_dir: str) -> dict:
if not os.path.isdir(templates_dir):
return {"templates": []}
templates: List[str] = []
for entry in os.listdir(templates_dir):
path = os.path.join(templates_dir, entry)
if os.path.isfile(path) and entry.lower().endswith(".docx"):
templates.append(entry)
templates.sort()
return {"templates": templates}
def open_template_path(templates_dir: str, name: str) -> str:
path = os.path.join(templates_dir, name)
if not os.path.isfile(path):
raise ValueError(f"Template not found: {name}")
return path
BIN
View File
Binary file not shown.