feat: add upload_attachment MCP tool
All checks were successful
Build and Push Docker Image / build (push) Successful in 24s

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.
This commit is contained in:
2026-01-03 19:59:47 -05:00
parent ea175d55a2
commit 848cfd684f
8 changed files with 292 additions and 5 deletions

View File

@@ -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

View File

@@ -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 = [

View File

@@ -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:

View File

@@ -22,6 +22,7 @@ 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
@@ -218,6 +219,32 @@ def create_server(
"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",
@@ -324,6 +351,12 @@ def create_server(
_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":

View File

@@ -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)

View File

@@ -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

View File

@@ -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

View File

@@ -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"
)