feat: add upload_attachment MCP tool

Add support for uploading file attachments to Grist documents:

- GristClient.upload_attachment() method using multipart/form-data
- upload_attachment tool function with base64 decoding and MIME detection
- Tool registration in server.py
- Comprehensive unit tests (7 new tests)

Returns attachment ID for linking to records via update_records.

Bumps version to 1.3.0.

.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,32 @@ 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
+- **`upload_attachment` MCP tool**: Upload files to Grist documents
+- Base64-encoded content input (required for JSON-based MCP protocol)
+- Automatic MIME type detection from filename
+- Returns attachment ID for linking to records via `update_records`
+
+#### Usage
+```python
+# 1. Upload attachment
+result = upload_attachment(
+    document="accounting",
+    filename="invoice.pdf",
+    content_base64="JVBERi0xLjQK..."
+)
+# Returns: {"attachment_id": 42, "filename": "invoice.pdf", "size_bytes": 31395}
+
+# 2. Link to record
+update_records(document="accounting", table="Bills", records=[
+    {"id": 1, "fields": {"Attachment": [42]}}
+])
+```
+
 ## [1.2.0] - 2026-01-02
 
 ### Added

README.md

@@ -150,6 +150,7 @@ Add to your MCP client configuration (e.g., Claude Desktop):
 | `GRIST_MCP_TOKEN` | Agent authentication token (required) | - |
 | `CONFIG_PATH` | Path to config file inside container | `/app/config.yaml` |
 | `LOG_LEVEL` | Logging verbosity (`DEBUG`, `INFO`, `WARNING`, `ERROR`) | `INFO` |
+| `GRIST_MCP_URL` | Public URL of this server (for session proxy tokens) | - |
 
 ### config.yaml Structure
 

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

@@ -127,6 +127,7 @@ def create_app(config: Config):
     """Create the ASGI application."""
     auth = Authenticator(config)
     token_manager = SessionTokenManager()
+    proxy_base_url = os.environ.get("GRIST_MCP_URL")
 
     sse = SseServerTransport("/messages")
 
@@ -144,7 +145,7 @@ def create_app(config: Config):
             return
 
         # Create a server instance for this authenticated connection
-        server = create_server(auth, agent, token_manager)
+        server = create_server(auth, agent, token_manager, proxy_base_url)
 
         async with sse.connect_sse(scope, receive, send) as streams:
             await server.run(
@@ -251,11 +252,18 @@ def create_app(config: Config):
 
 def _print_mcp_config(external_port: int, tokens: list) -> None:
     """Print Claude Code MCP configuration."""
+    # Use GRIST_MCP_URL if set, otherwise fall back to localhost
+    base_url = os.environ.get("GRIST_MCP_URL")
+    if base_url:
+        sse_url = f"{base_url.rstrip('/')}/sse"
+    else:
+        sse_url = f"http://localhost:{external_port}/sse"
+
     print()
     print("Claude Code MCP configuration (copy-paste to add):")
     for t in tokens:
         config = (
-            f'{{"type": "sse", "url": "http://localhost:{external_port}/sse", '
+            f'{{"type": "sse", "url": "{sse_url}", '
             f'"headers": {{"Authorization": "Bearer {t.token}"}}}}'
         )
         print(f"  claude mcp add-json grist-{t.name} '{config}'")

src/grist_mcp/server.py

@@ -22,25 +22,33 @@ from grist_mcp.tools.read import sql_query as _sql_query
 from grist_mcp.tools.write import add_records as _add_records
 from grist_mcp.tools.write import update_records as _update_records
 from grist_mcp.tools.write import delete_records as _delete_records
+from grist_mcp.tools.write import upload_attachment as _upload_attachment
 from grist_mcp.tools.schema import create_table as _create_table
 from grist_mcp.tools.schema import add_column as _add_column
 from grist_mcp.tools.schema import modify_column as _modify_column
 from grist_mcp.tools.schema import delete_column as _delete_column
 
 
-def create_server(auth: Authenticator, agent: Agent, token_manager: SessionTokenManager | None = None) -> Server:
+def create_server(
+    auth: Authenticator,
+    agent: Agent,
+    token_manager: SessionTokenManager | None = None,
+    proxy_base_url: str | None = None,
+) -> Server:
     """Create and configure the MCP server for an authenticated agent.
 
     Args:
         auth: Authenticator instance for permission checks.
         agent: The authenticated agent for this server instance.
         token_manager: Optional session token manager for HTTP proxy access.
+        proxy_base_url: Base URL for the proxy endpoint (e.g., "https://example.com").
 
     Returns:
         Configured MCP Server instance.
     """
     server = Server("grist-mcp")
     _current_agent = agent
+    _proxy_base_url = proxy_base_url
 
     @server.list_tools()
     async def list_tools() -> list[Tool]:
@@ -211,6 +219,32 @@ def create_server(auth: Authenticator, agent: Agent, token_manager: SessionToken
                     "required": ["document", "table", "column_id"],
                 },
             ),
+            Tool(
+                name="upload_attachment",
+                description="Upload a file attachment to a Grist document. Returns attachment ID for linking to records via update_records.",
+                inputSchema={
+                    "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)",
+                        },
+                    },
+                    "required": ["document", "filename", "content_base64"],
+                },
+            ),
             Tool(
                 name="get_proxy_documentation",
                 description="Get complete documentation for the HTTP proxy API",
@@ -317,6 +351,12 @@ def create_server(auth: Authenticator, agent: Agent, token_manager: SessionToken
                     _current_agent, auth, arguments["document"], arguments["table"],
                     arguments["column_id"],
                 )
+            elif name == "upload_attachment":
+                result = await _upload_attachment(
+                    _current_agent, auth, arguments["document"],
+                    arguments["filename"], arguments["content_base64"],
+                    content_type=arguments.get("content_type"),
+                )
             elif name == "get_proxy_documentation":
                 result = await _get_proxy_documentation()
             elif name == "request_session_token":
@@ -327,6 +367,7 @@ def create_server(auth: Authenticator, agent: Agent, token_manager: SessionToken
                     arguments["document"],
                     arguments["permissions"],
                     ttl_seconds=arguments.get("ttl_seconds", 300),
+                    proxy_base_url=_proxy_base_url,
                 )
             else:
                 return [TextContent(type="text", text=f"Unknown tool: {name}")]

src/grist_mcp/tools/session.py

@@ -7,6 +7,7 @@ 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",
     "authentication": "Bearer token in Authorization header",
     "request_format": {
         "method": "Operation name (required)",
@@ -88,11 +89,12 @@ PROXY_DOCUMENTATION = {
 import requests
 import sys
 
+# Use token and proxy_url from request_session_token response
 token = sys.argv[1]
-host = sys.argv[2]
+proxy_url = sys.argv[2]
 
 response = requests.post(
-    f'{host}/api/v1/proxy',
+    proxy_url,
     headers={'Authorization': f'Bearer {token}'},
     json={
         'method': 'add_records',
@@ -117,6 +119,7 @@ async def request_session_token(
     document: str,
     permissions: list[str],
     ttl_seconds: int = 300,
+    proxy_base_url: str | None = None,
 ) -> dict:
     """Request a short-lived session token for HTTP proxy access.
 
@@ -139,10 +142,17 @@ async def request_session_token(
         ttl_seconds=ttl_seconds,
     )
 
+    # Build proxy URL - use base URL if provided, otherwise just path
+    proxy_path = "/api/v1/proxy"
+    if proxy_base_url:
+        proxy_url = f"{proxy_base_url.rstrip('/')}{proxy_path}"
+    else:
+        proxy_url = proxy_path
+
     return {
         "token": session.token,
         "document": session.document,
         "permissions": session.permissions,
         "expires_at": session.expires_at.isoformat(),
-        "proxy_url": "/api/v1/proxy",
+        "proxy_url": proxy_url,
     }

src/grist_mcp/tools/write.py

@@ -1,4 +1,7 @@
-"""Write tools - create, update, delete records."""
+"""Write tools - create, update, delete records, upload attachments."""
+
+import base64
+import mimetypes
 
 from grist_mcp.auth import Agent, Authenticator, Permission
 from grist_mcp.grist_client import GristClient
@@ -59,3 +62,50 @@ async def delete_records(
 
     await client.delete_records(table, record_ids)
     return {"deleted": True}
+
+
+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.
+
+    Args:
+        agent: The authenticated agent.
+        auth: Authenticator for permission checks.
+        document: Document name.
+        filename: Filename with extension.
+        content_base64: File content as base64-encoded string.
+        content_type: MIME type (auto-detected from filename if omitted).
+        client: Optional GristClient instance.
+
+    Returns:
+        Dict with attachment_id, filename, and size_bytes.
+
+    Raises:
+        ValueError: If content_base64 is not valid base64.
+    """
+    auth.authorize(agent, document, Permission.WRITE)
+
+    # Decode base64 content
+    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)

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_server.py

@@ -52,13 +52,14 @@ tokens:
     assert "add_column" in tool_names
     assert "modify_column" in tool_names
     assert "delete_column" in tool_names
+    assert "upload_attachment" in tool_names
 
     # Session tools (always registered)
     assert "get_proxy_documentation" in tool_names
     assert "request_session_token" in tool_names
 
-    # Should have all 14 tools
-    assert len(result.root.tools) == 14
+    # Should have all 15 tools
+    assert len(result.root.tools) == 15
 
 
 @pytest.mark.asyncio

tests/unit/test_tools_write.py

@@ -1,7 +1,9 @@
+import base64
+
 import pytest
 from unittest.mock import AsyncMock
 
-from grist_mcp.tools.write import add_records, update_records, delete_records
+from grist_mcp.tools.write import add_records, update_records, delete_records, upload_attachment
 from grist_mcp.auth import Authenticator, AuthError
 from grist_mcp.config import Config, Document, Token, TokenScope
 
@@ -94,3 +96,105 @@ async def test_delete_records(auth, mock_client):
     )
 
     assert result == {"deleted": True}
+
+
+# Upload attachment tests
+
+@pytest.fixture
+def mock_client_with_attachment():
+    client = AsyncMock()
+    client.upload_attachment.return_value = {
+        "attachment_id": 42,
+        "filename": "invoice.pdf",
+        "size_bytes": 1024,
+    }
+    return client
+
+
+@pytest.mark.asyncio
+async def test_upload_attachment_success(auth, mock_client_with_attachment):
+    agent = auth.authenticate("write-token")
+    content = b"PDF content"
+    content_base64 = base64.b64encode(content).decode()
+
+    result = await upload_attachment(
+        agent, auth, "budget",
+        filename="invoice.pdf",
+        content_base64=content_base64,
+        client=mock_client_with_attachment,
+    )
+
+    assert result == {
+        "attachment_id": 42,
+        "filename": "invoice.pdf",
+        "size_bytes": 1024,
+    }
+    mock_client_with_attachment.upload_attachment.assert_called_once_with(
+        "invoice.pdf", content, "application/pdf"
+    )
+
+
+@pytest.mark.asyncio
+async def test_upload_attachment_invalid_base64(auth, mock_client_with_attachment):
+    agent = auth.authenticate("write-token")
+
+    with pytest.raises(ValueError, match="Invalid base64 encoding"):
+        await upload_attachment(
+            agent, auth, "budget",
+            filename="test.txt",
+            content_base64="not-valid-base64!!!",
+            client=mock_client_with_attachment,
+        )
+
+
+@pytest.mark.asyncio
+async def test_upload_attachment_auth_required(auth, mock_client_with_attachment):
+    agent = auth.authenticate("read-token")
+    content_base64 = base64.b64encode(b"test").decode()
+
+    with pytest.raises(AuthError, match="Permission denied"):
+        await upload_attachment(
+            agent, auth, "budget",
+            filename="test.txt",
+            content_base64=content_base64,
+            client=mock_client_with_attachment,
+        )
+
+
+@pytest.mark.asyncio
+async def test_upload_attachment_mime_detection(auth, mock_client_with_attachment):
+    agent = auth.authenticate("write-token")
+    content = b"PNG content"
+    content_base64 = base64.b64encode(content).decode()
+
+    await upload_attachment(
+        agent, auth, "budget",
+        filename="image.png",
+        content_base64=content_base64,
+        client=mock_client_with_attachment,
+    )
+
+    # Should auto-detect image/png from filename
+    mock_client_with_attachment.upload_attachment.assert_called_once_with(
+        "image.png", content, "image/png"
+    )
+
+
+@pytest.mark.asyncio
+async def test_upload_attachment_explicit_content_type(auth, mock_client_with_attachment):
+    agent = auth.authenticate("write-token")
+    content = b"custom content"
+    content_base64 = base64.b64encode(content).decode()
+
+    await upload_attachment(
+        agent, auth, "budget",
+        filename="file.dat",
+        content_base64=content_base64,
+        content_type="application/custom",
+        client=mock_client_with_attachment,
+    )
+
+    # Should use explicit content type
+    mock_client_with_attachment.upload_attachment.assert_called_once_with(
+        "file.dat", content, "application/custom"
+    )

uv.lock

@@ -153,7 +153,7 @@ wheels = [
 
 [[package]]
 name = "grist-mcp"
-version = "1.1.0"
+version = "1.2.0"
 source = { editable = "." }
 dependencies = [
     { name = "httpx" },