feat: replace MCP attachment tool with proxy endpoint

The MCP tool approach was impractical because it required the LLM to
generate large base64 strings token-by-token, causing timeouts.

Changes:
- Remove upload_attachment MCP tool
- Add POST /api/v1/attachments endpoint for multipart/form-data uploads
- Update proxy documentation to show both endpoints
- Uses existing GristClient.upload_attachment() method
- Requires write permission in session token

.github/workflows/build.yaml

@@ -18,7 +18,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6
 
       - name: Log in to Container Registry
         uses: docker/login-action@v3

CHANGELOG.md

@@ -5,6 +5,50 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.3.0] - 2026-01-03
+
+### Added
+
+#### Attachment Upload via Proxy
+- **`POST /api/v1/attachments`**: New HTTP endpoint for file uploads
+- Uses `multipart/form-data` for efficient binary transfer (no base64 overhead)
+- Automatic MIME type detection from filename
+- Returns attachment ID for linking to records via `update_records`
+- Requires write permission in session token
+
+#### Usage
+```bash
+# Get session token with write permission
+TOKEN=$(curl -s ... | jq -r '.token')
+
+# Upload file
+curl -X POST \
+  -H "Authorization: Bearer $TOKEN" \
+  -F "file=@invoice.pdf" \
+  https://example.com/api/v1/attachments
+
+# Returns: {"success": true, "data": {"attachment_id": 42, "filename": "invoice.pdf", "size_bytes": 31395}}
+```
+
+```python
+# Python example
+import requests
+
+response = requests.post(
+    f'{proxy_url.replace("/proxy", "/attachments")}',
+    headers={'Authorization': f'Bearer {token}'},
+    files={'file': open('invoice.pdf', 'rb')}
+)
+attachment_id = response.json()['data']['attachment_id']
+
+# Link to record via proxy
+requests.post(proxy_url, headers={'Authorization': f'Bearer {token}'}, json={
+    'method': 'update_records',
+    'table': 'Bills',
+    'records': [{'id': 1, 'fields': {'Attachment': [attachment_id]}}]
+})
+```
+
 ## [1.2.0] - 2026-01-02
 
 ### Added

docs/plans/2026-01-03-attachment-upload-design.md

@@ -0,0 +1,187 @@
+# Attachment Upload Feature Design
+
+**Date:** 2026-01-03
+**Status:** Approved
+
+## Summary
+
+Add an `upload_attachment` MCP tool to upload files to Grist documents and receive an attachment ID for linking to records.
+
+## Design Decisions
+
+| Decision | Choice | Rationale |
+|----------|--------|-----------|
+| Content encoding | Base64 string | MCP tools use JSON; binary must be encoded |
+| Batch support | Single file only | YAGNI; caller can loop if needed |
+| Linking behavior | Upload only, return ID | Single responsibility; use existing `update_records` to link |
+| Download support | Not included | YAGNI; can add later if needed |
+| Permission level | Write | Attachments are data, not schema |
+| Proxy support | MCP tool only | Reduces scope; scripts can use Grist API directly |
+
+## Tool Interface
+
+### Input Schema
+
+```json
+{
+  "type": "object",
+  "properties": {
+    "document": {
+      "type": "string",
+      "description": "Document name"
+    },
+    "filename": {
+      "type": "string",
+      "description": "Filename with extension (e.g., 'invoice.pdf')"
+    },
+    "content_base64": {
+      "type": "string",
+      "description": "File content as base64-encoded string"
+    },
+    "content_type": {
+      "type": "string",
+      "description": "MIME type (optional, auto-detected from filename if omitted)"
+    }
+  },
+  "required": ["document", "filename", "content_base64"]
+}
+```
+
+### Response
+
+```json
+{
+  "attachment_id": 42,
+  "filename": "invoice.pdf",
+  "size_bytes": 30720
+}
+```
+
+### Usage Example
+
+```python
+# 1. Upload attachment
+result = upload_attachment(
+    document="accounting",
+    filename="Invoice-001.pdf",
+    content_base64="JVBERi0xLjQK..."
+)
+
+# 2. Link to record via existing update_records tool
+update_records("Bills", [{
+    "id": 1,
+    "fields": {"Attachment": [result["attachment_id"]]}
+}])
+```
+
+## Implementation
+
+### Files to Modify
+
+1. **`src/grist_mcp/grist_client.py`** - Add `upload_attachment()` method
+2. **`src/grist_mcp/tools/write.py`** - Add tool function
+3. **`src/grist_mcp/server.py`** - Register tool
+
+### GristClient Method
+
+```python
+async def upload_attachment(
+    self,
+    filename: str,
+    content: bytes,
+    content_type: str | None = None
+) -> dict:
+    """Upload a file attachment. Returns attachment metadata."""
+    if content_type is None:
+        content_type = "application/octet-stream"
+
+    files = {"upload": (filename, content, content_type)}
+
+    async with httpx.AsyncClient(timeout=self._timeout) as client:
+        response = await client.post(
+            f"{self._base_url}/attachments",
+            headers=self._headers,
+            files=files,
+        )
+        response.raise_for_status()
+        # Grist returns list of attachment IDs
+        attachment_ids = response.json()
+        return {
+            "attachment_id": attachment_ids[0],
+            "filename": filename,
+            "size_bytes": len(content),
+        }
+```
+
+### Tool Function
+
+```python
+import base64
+import mimetypes
+
+async def upload_attachment(
+    agent: Agent,
+    auth: Authenticator,
+    document: str,
+    filename: str,
+    content_base64: str,
+    content_type: str | None = None,
+    client: GristClient | None = None,
+) -> dict:
+    """Upload a file attachment to a document."""
+    auth.authorize(agent, document, Permission.WRITE)
+
+    # Decode base64
+    try:
+        content = base64.b64decode(content_base64)
+    except Exception:
+        raise ValueError("Invalid base64 encoding")
+
+    # Auto-detect MIME type if not provided
+    if content_type is None:
+        content_type, _ = mimetypes.guess_type(filename)
+        if content_type is None:
+            content_type = "application/octet-stream"
+
+    if client is None:
+        doc = auth.get_document(document)
+        client = GristClient(doc)
+
+    return await client.upload_attachment(filename, content, content_type)
+```
+
+## Error Handling
+
+| Error | Cause | Response |
+|-------|-------|----------|
+| Invalid base64 | Malformed content_base64 | `ValueError: Invalid base64 encoding` |
+| Authorization | Agent lacks write permission | `AuthError` (existing pattern) |
+| Grist API error | Upload fails | `httpx.HTTPStatusError` (existing pattern) |
+
+## Testing
+
+### Unit Tests
+
+**`tests/unit/test_tools_write.py`:**
+- `test_upload_attachment_success` - Valid base64, returns attachment_id
+- `test_upload_attachment_invalid_base64` - Raises ValueError
+- `test_upload_attachment_auth_required` - Verifies write permission check
+- `test_upload_attachment_mime_detection` - Auto-detects type from filename
+
+**`tests/unit/test_grist_client.py`:**
+- `test_upload_attachment_api_call` - Correct multipart request format
+- `test_upload_attachment_with_explicit_content_type` - Passes through MIME type
+
+### Mock Approach
+
+Mock `httpx.AsyncClient` responses; no Grist server needed for unit tests.
+
+## Future Considerations
+
+Not included in this implementation (YAGNI):
+- Batch upload (multiple files)
+- Download attachment
+- Proxy API support
+- Size limit validation (rely on Grist's limits)
+
+These can be added if real use cases emerge.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "grist-mcp"
-version = "1.2.0"
+version = "1.3.0"
 description = "MCP server for AI agents to interact with Grist documents"
 requires-python = ">=3.14"
 dependencies = [

src/grist_mcp/grist_client.py

@@ -116,6 +116,39 @@ class GristClient:
         """Delete records by ID."""
         await self._request("POST", f"/tables/{table}/data/delete", json=record_ids)
 
+    async def upload_attachment(
+        self,
+        filename: str,
+        content: bytes,
+        content_type: str = "application/octet-stream",
+    ) -> dict:
+        """Upload a file attachment. Returns attachment metadata.
+
+        Args:
+            filename: Name for the uploaded file.
+            content: File content as bytes.
+            content_type: MIME type of the file.
+
+        Returns:
+            Dict with attachment_id, filename, and size_bytes.
+        """
+        files = {"upload": (filename, content, content_type)}
+
+        async with httpx.AsyncClient(timeout=self._timeout) as client:
+            response = await client.post(
+                f"{self._base_url}/attachments",
+                headers=self._headers,
+                files=files,
+            )
+            response.raise_for_status()
+            # Grist returns list of attachment IDs
+            attachment_ids = response.json()
+            return {
+                "attachment_id": attachment_ids[0],
+                "filename": filename,
+                "size_bytes": len(content),
+            }
+
     # Schema operations
 
     async def create_table(self, table_id: str, columns: list[dict]) -> str:

src/grist_mcp/main.py

@@ -14,6 +14,7 @@ from grist_mcp.config import Config, load_config
 from grist_mcp.auth import Authenticator, AuthError
 from grist_mcp.session import SessionTokenManager
 from grist_mcp.proxy import parse_proxy_request, dispatch_proxy_request, ProxyError
+from grist_mcp.grist_client import GristClient
 from grist_mcp.logging import setup_logging
 
 
@@ -59,6 +60,62 @@ async def send_json_response(send: Send, status: int, data: dict) -> None:
     })
 
 
+def _parse_multipart(content_type: str, body: bytes) -> tuple[str | None, bytes | None]:
+    """Parse multipart/form-data to extract uploaded file.
+
+    Returns (filename, content) or (None, None) if parsing fails.
+    """
+    import re
+
+    # Extract boundary from content-type
+    match = re.search(r'boundary=([^\s;]+)', content_type)
+    if not match:
+        return None, None
+
+    boundary = match.group(1).encode()
+    if boundary.startswith(b'"') and boundary.endswith(b'"'):
+        boundary = boundary[1:-1]
+
+    # Split by boundary
+    parts = body.split(b'--' + boundary)
+
+    for part in parts:
+        if b'Content-Disposition' not in part:
+            continue
+
+        # Split headers from content
+        if b'\r\n\r\n' in part:
+            header_section, content = part.split(b'\r\n\r\n', 1)
+        elif b'\n\n' in part:
+            header_section, content = part.split(b'\n\n', 1)
+        else:
+            continue
+
+        headers = header_section.decode('utf-8', errors='replace')
+
+        # Check if this is a file upload
+        if 'filename=' not in headers:
+            continue
+
+        # Extract filename
+        filename_match = re.search(r'filename="([^"]+)"', headers)
+        if not filename_match:
+            filename_match = re.search(r"filename=([^\s;]+)", headers)
+        if not filename_match:
+            continue
+
+        filename = filename_match.group(1)
+
+        # Remove trailing boundary marker and whitespace
+        content = content.rstrip()
+        if content.endswith(b'--'):
+            content = content[:-2].rstrip()
+
+        return filename, content
+
+    return None, None
+
+
 CONFIG_TEMPLATE = """\
 # grist-mcp configuration
 #
@@ -229,6 +286,83 @@ def create_app(config: Config):
                 "code": e.code,
             })
 
+    async def handle_attachments(scope: Scope, receive: Receive, send: Send) -> None:
+        """Handle file attachment uploads via multipart/form-data."""
+        # Extract token
+        token = _get_bearer_token(scope)
+        if not token:
+            await send_json_response(send, 401, {
+                "success": False,
+                "error": "Missing Authorization header",
+                "code": "INVALID_TOKEN",
+            })
+            return
+
+        # Validate session token
+        session = token_manager.validate_token(token)
+        if session is None:
+            await send_json_response(send, 401, {
+                "success": False,
+                "error": "Invalid or expired token",
+                "code": "TOKEN_EXPIRED",
+            })
+            return
+
+        # Check write permission
+        if "write" not in session.permissions:
+            await send_json_response(send, 403, {
+                "success": False,
+                "error": "Write permission required for attachment upload",
+                "code": "UNAUTHORIZED",
+            })
+            return
+
+        # Get content-type header
+        headers = dict(scope.get("headers", []))
+        content_type = headers.get(b"content-type", b"").decode()
+
+        if not content_type.startswith("multipart/form-data"):
+            await send_json_response(send, 400, {
+                "success": False,
+                "error": "Content-Type must be multipart/form-data",
+                "code": "INVALID_REQUEST",
+            })
+            return
+
+        # Read request body
+        body = b""
+        while True:
+            message = await receive()
+            body += message.get("body", b"")
+            if not message.get("more_body", False):
+                break
+
+        # Parse multipart
+        filename, content = _parse_multipart(content_type, body)
+        if filename is None or content is None:
+            await send_json_response(send, 400, {
+                "success": False,
+                "error": "No file found in request",
+                "code": "INVALID_REQUEST",
+            })
+            return
+
+        # Upload to Grist
+        try:
+            doc = auth.get_document(session.document)
+            client = GristClient(doc)
+            result = await client.upload_attachment(filename, content)
+            await send_json_response(send, 200, {
+                "success": True,
+                "data": result,
+            })
+        except Exception as e:
+            await send_json_response(send, 500, {
+                "success": False,
+                "error": str(e),
+                "code": "GRIST_ERROR",
+            })
+
     async def app(scope: Scope, receive: Receive, send: Send) -> None:
         if scope["type"] != "http":
             return
@@ -244,6 +378,8 @@ def create_app(config: Config):
             await handle_messages(scope, receive, send)
         elif path == "/api/v1/proxy" and method == "POST":
             await handle_proxy(scope, receive, send)
+        elif path == "/api/v1/attachments" and method == "POST":
+            await handle_attachments(scope, receive, send)
         else:
             await handle_not_found(scope, receive, send)
 

src/grist_mcp/tools/session.py

@@ -6,9 +6,28 @@ from grist_mcp.session import SessionTokenManager
 
 PROXY_DOCUMENTATION = {
     "description": "HTTP proxy API for bulk data operations. Use request_session_token to get a short-lived token, then call the proxy endpoint directly from scripts.",
-    "endpoint": "POST /api/v1/proxy",
-    "endpoint_note": "The full URL is returned in the 'proxy_url' field of request_session_token response",
+    "endpoints": {
+        "proxy": "POST /api/v1/proxy - JSON operations (CRUD, schema)",
+        "attachments": "POST /api/v1/attachments - File uploads (multipart/form-data)",
+    },
+    "endpoint_note": "The full URL is returned in the 'proxy_url' field of request_session_token response. Replace /proxy with /attachments for file uploads.",
     "authentication": "Bearer token in Authorization header",
+    "attachment_upload": {
+        "endpoint": "POST /api/v1/attachments",
+        "content_type": "multipart/form-data",
+        "permission": "write",
+        "description": "Upload file attachments to the document. Returns attachment_id for linking to records via update_records.",
+        "response": {"success": True, "data": {"attachment_id": 42, "filename": "invoice.pdf", "size_bytes": 31395}},
+        "example_curl": "curl -X POST -H 'Authorization: Bearer TOKEN' -F 'file=@invoice.pdf' URL/api/v1/attachments",
+        "example_python": """import requests
+response = requests.post(
+    f'{proxy_url.replace("/proxy", "/attachments")}',
+    headers={'Authorization': f'Bearer {token}'},
+    files={'file': open('invoice.pdf', 'rb')}
+)
+attachment_id = response.json()['data']['attachment_id']
+# Link to record: update_records with {'Attachment': [attachment_id]}""",
+    },
     "request_format": {
         "method": "Operation name (required)",
         "table": "Table name (required for most operations)",

tests/unit/test_grist_client.py

@@ -196,3 +196,43 @@ def test_sql_validation_rejects_multiple_statements(client):
 def test_sql_validation_allows_trailing_semicolon(client):
     # Should not raise
     client._validate_sql_query("SELECT * FROM users;")
+
+
+# Attachment tests
+
+@pytest.mark.asyncio
+async def test_upload_attachment(client, httpx_mock: HTTPXMock):
+    httpx_mock.add_response(
+        url="https://grist.example.com/api/docs/abc123/attachments",
+        method="POST",
+        json=[42],
+    )
+
+    result = await client.upload_attachment(
+        filename="invoice.pdf",
+        content=b"PDF content here",
+        content_type="application/pdf",
+    )
+
+    assert result == {
+        "attachment_id": 42,
+        "filename": "invoice.pdf",
+        "size_bytes": 16,
+    }
+
+
+@pytest.mark.asyncio
+async def test_upload_attachment_default_content_type(client, httpx_mock: HTTPXMock):
+    httpx_mock.add_response(
+        url="https://grist.example.com/api/docs/abc123/attachments",
+        method="POST",
+        json=[99],
+    )
+
+    result = await client.upload_attachment(
+        filename="data.bin",
+        content=b"\x00\x01\x02",
+    )
+
+    assert result["attachment_id"] == 99
+    assert result["size_bytes"] == 3

tests/unit/test_tools_session.py

@@ -39,12 +39,14 @@ async def test_get_proxy_documentation_returns_complete_spec():
     result = await get_proxy_documentation()
 
     assert "description" in result
-    assert "endpoint" in result
-    assert result["endpoint"] == "POST /api/v1/proxy"
+    assert "endpoints" in result
+    assert "proxy" in result["endpoints"]
+    assert "attachments" in result["endpoints"]
     assert "authentication" in result
     assert "methods" in result
     assert "add_records" in result["methods"]
     assert "get_records" in result["methods"]
+    assert "attachment_upload" in result
     assert "example_script" in result