Python API reference

These sections document the existing server modules from their Python docstrings. The MCP-facing read_emails function is registered inside a helper; its client contract is documented explicitly in the tool reference.

The pre-push hook builds this reference with strict link and anchor checks. Use make docs-build for earlier feedback when needed; do not repeat a passing hook gate manually.

Configuration store

Read-only file-backed store for integrations.json.

IntegrationsFileError

Bases: Exception

Raised when integrations.json is missing, empty, or corrupt.

Source code in src/justpen_integration_mcp/store.py

class IntegrationsFileError(Exception):
    """Raised when integrations.json is missing, empty, or corrupt."""

get_integrations_path()

Return the configured integrations path or the container default.

Source code in src/justpen_integration_mcp/store.py

def get_integrations_path() -> Path:
    """Return the configured integrations path or the container default."""
    if _State.path is not None:
        return _State.path
    return Path("/justpen_home") / "integrations.json"

load() async

Load and parse integrations.json. Read-only — no writes.

Source code in src/justpen_integration_mcp/store.py

async def load() -> dict[str, Any]:
    """Load and parse integrations.json. Read-only — no writes."""
    path = get_integrations_path()
    if not path.exists():
        raise IntegrationsFileError(f"{path} not found")
    text = path.read_text()
    if not text.strip():
        raise IntegrationsFileError(f"{path} is empty")
    try:
        data = json.loads(text)
    except json.JSONDecodeError as e:
        raise IntegrationsFileError(f"Invalid JSON in {path}: {e}") from e
    if not isinstance(data, dict):
        raise IntegrationsFileError(f"integrations.json root must be a dict, got {type(data).__name__}")
    # json.loads returns Any; narrowing via isinstance only tells pyright it is a dict with unknown
    # key/value types. Cast to dict[str, Any] — JSON object keys are always strings by spec.
    return cast("dict[str, Any]", data)

load_email_config() async

Load and validate the email block from integrations.json.

Returns:

Type	Description
dict[str, Any]	{"service_account_key": str, "delegated_user": str}

Raises:

Type	Description
IntegrationsFileError	On any config problem.

Source code in src/justpen_integration_mcp/store.py

async def load_email_config() -> dict[str, Any]:
    """Load and validate the email block from integrations.json.

    Returns:
        {"service_account_key": str, "delegated_user": str}

    Raises:
        IntegrationsFileError: On any config problem.
    """
    data = await load()
    email_raw_any = data.get("email")
    if not email_raw_any or not isinstance(email_raw_any, dict):
        raise IntegrationsFileError("'email' not configured in integrations.json")
    # Narrow dict[Unknown, Unknown] from isinstance to dict[str, Any]: JSON object keys are strings by spec.
    email_raw = cast("dict[str, Any]", email_raw_any)

    email_type = email_raw.get("type")
    if email_type != "gmail":
        raise IntegrationsFileError(f"Unsupported email type '{email_type}' (only 'gmail' supported)")

    key_path_raw = email_raw.get("service_account_json_path", "")
    if not isinstance(key_path_raw, str) or not key_path_raw:
        raise IntegrationsFileError(f"service_account_json_path not found: '{key_path_raw}'")
    key_path: str = key_path_raw
    resolved_key_path = await asyncio.to_thread(lambda: Path(key_path).expanduser())
    key_exists = await asyncio.to_thread(resolved_key_path.is_file)
    if not key_exists:
        raise IntegrationsFileError(f"service_account_json_path not found: '{key_path}'")

    delegated_user_raw = email_raw.get("delegated_user", "")
    if not isinstance(delegated_user_raw, str):
        raise IntegrationsFileError("delegated_user must be a string")
    delegated_user = delegated_user_raw.strip()
    if not delegated_user:
        raise IntegrationsFileError("delegated_user must not be empty")

    return {
        "service_account_key": str(resolved_key_path),
        "delegated_user": delegated_user,
    }

set_integrations_path(path)

Set the path used to read the integrations configuration.

Source code in src/justpen_integration_mcp/store.py

def set_integrations_path(path: Path) -> None:
    """Set the path used to read the integrations configuration."""
    _State.path = path

Response helpers

Successful tool responses use status: "ok"; failures use status: "error" and an error string. The tool reference shows the data shapes returned by read_emails.

Response builders for integration MCP tools.

error_response(error)

Build an error response containing the supplied message.

Source code in src/justpen_integration_mcp/responses.py

def error_response(error: str) -> dict[str, Any]:
    """Build an error response containing the supplied message."""
    return {"status": "error", "error": error}

ok_response(data=None)

Build an ok response, including data only when provided.

Source code in src/justpen_integration_mcp/responses.py

def ok_response(data: dict[str, Any] | list[Any] | None = None) -> dict[str, Any]:
    """Build an ok response, including data only when provided."""
    r: dict[str, Any] = {"status": "ok"}
    if data is not None:
        r["data"] = data
    return r

Email registration

Email reading tool — Gmail via service account delegation.

register(mcp)

Register the Gmail email-reading tool on the MCP server.

Source code in src/justpen_integration_mcp/tools/email.py

def register(mcp: FastMCP) -> None:
    """Register the Gmail email-reading tool on the MCP server."""
    _register_read_emails(mcp)