SafetyCulture MCP Server Implementation Plan#

[project]
name = "safetyculture-mcp"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "fastmcp==3.4.2",
    "httpx>=0.27.0",
    "pydantic>=2.0.0",
    "python-dotenv",
]

fastmcp==3.4.2
httpx>=0.27.0
pydantic>=2.0.0
python-dotenv

pytest>=8.0.0
pytest-asyncio>=0.23.0

SAFETYCULTURE_API_TOKEN=your_token_here

.env
.venv/
__pycache__/
*.pyc
.pytest_cache/
*.egg-info/
dist/
build/

import os
os.environ.setdefault("SAFETYCULTURE_API_TOKEN", "test_token_for_testing")

.venv\Scripts\Activate.ps1
pip install --upgrade pip
pip install -r requirements.txt
pip install -r requirements-dev.txt
pip install -e .

fastmcp version

FastMCP version: 3.4.2
MCP version: 1.27.2
Python version: 3.12.x

git init
git add pyproject.toml requirements.txt requirements-dev.txt .env.example .gitignore src/ tests/
git commit -m "chore: scaffold safetyculture-mcp project structure"

import pytest
from safetyculture_mcp.client import BASE_URL, HEADERS, raise_for_status
import httpx
from fastmcp.exceptions import ToolError

def test_client_base_url():
    assert BASE_URL == "https://api.safetyculture.io"

def test_client_headers_has_authorization():
    assert "Authorization" in HEADERS
    assert HEADERS["Authorization"].startswith("Bearer ")

def test_client_headers_has_content_type():
    assert HEADERS["Content-Type"] == "application/json"

def test_raise_for_status_401():
    resp = httpx.Response(401, request=httpx.Request("GET", "http://test"))
    with pytest.raises(ToolError, match="Invalid or expired"):
        raise_for_status(resp)

def test_raise_for_status_404():
    resp = httpx.Response(404, request=httpx.Request("GET", "http://test"))
    with pytest.raises(ToolError, match="not found"):
        raise_for_status(resp)

def test_raise_for_status_429():
    resp = httpx.Response(429, request=httpx.Request("GET", "http://test"))
    with pytest.raises(ToolError, match="rate limit"):
        raise_for_status(resp)

def test_raise_for_status_500():
    resp = httpx.Response(500, request=httpx.Request("GET", "http://test"))
    with pytest.raises(ToolError, match="server error"):
        raise_for_status(resp)

def test_raise_for_status_200_does_not_raise():
    resp = httpx.Response(200, request=httpx.Request("GET", "http://test"))
    raise_for_status(resp) # should not raise

pytest tests/test_tools.py -v -k "test_client"

import os
import httpx
from dotenv import load_dotenv
from fastmcp.exceptions import ToolError

load_dotenv()

API_TOKEN = os.environ["SAFETYCULTURE_API_TOKEN"]
BASE_URL = "https://api.safetyculture.io"
HEADERS = {
    "Authorization": f"Bearer {API_TOKEN}",
    "Content-Type": "application/json",
}

def raise_for_status(resp: httpx.Response) -> None:
    if resp.status_code == 401:
        raise ToolError("Invalid or expired SAFETYCULTURE_API_TOKEN. Check your .env file.")
    if resp.status_code == 404:
        raise ToolError(f"Resource not found: {resp.url}")
    if resp.status_code == 429:
        raise ToolError("SafetyCulture rate limit exceeded. Retry after the reset window.")
    if resp.status_code >= 500:
        raise ToolError(f"SafetyCulture server error ({resp.status_code}). Try again later.")
    resp.raise_for_status()

pytest tests/test_tools.py -v -k "test_client"

python -c "
import asyncio, httpx
from safetyculture_mcp.client import BASE_URL, HEADERS

async def check():
    async with httpx.AsyncClient() as client:
        resp = await client.get(f'{BASE_URL}/feed/v2/accounts/me', headers=HEADERS)
        print(resp.status_code, resp.json())

asyncio.run(check())
"

git add src/safetyculture_mcp/client.py tests/test_tools.py tests/conftest.py
git commit -m "feat: add client auth config and error helper"

from safetyculture_mcp.models.schemas import (
    InspectionSummary,
    InspectionDetail,
    Template,
    ActionTask,
    Action,
    CreatedAction,
    User,
)

def test_inspection_summary_required_field():
    s = InspectionSummary(audit_id="audit_1")
    assert s.audit_id == "audit_1"
    assert s.template_id is None
    assert s.modified_at is None

def test_inspection_detail_defaults():
    d = InspectionDetail(id="insp_1")
    assert d.id == "insp_1"
    assert d.is_marked_as_complete is False
    assert d.title is None

def test_template_required_field():
    t = Template(template_id="tmpl_1")
    assert t.template_id == "tmpl_1"
    assert t.name is None

def test_action_task_required_field():
    t = ActionTask(task_id="task_1")
    assert t.task_id == "task_1"
    assert t.status is None

def test_action_wraps_task():
    a = Action(task=ActionTask(task_id="task_1"))
    assert a.task.task_id == "task_1"

def test_created_action():
    ca = CreatedAction(action_id="act_1")
    assert ca.action_id == "act_1"

def test_user_defaults():
    u = User(id="user_1", email="a@b.com")
    assert u.active is True
    assert u.firstname is None

pytest tests/test_tools.py -v -k "test_inspection or test_template or test_action or test_user or test_created"

from pydantic import BaseModel

class InspectionSummary(BaseModel):
    audit_id: str
    template_id: str | None = None
    modified_at: str | None = None

class InspectionDetail(BaseModel):
    id: str
    template_id: str | None = None
    title: str | None = None
    created_at: str | None = None
    modified_at: str | None = None
    is_marked_as_complete: bool = False

class Template(BaseModel):
    template_id: str
    name: str | None = None
    modified_at: str | None = None
    created_at: str | None = None

class ActionTask(BaseModel):
    task_id: str
    title: str | None = None
    description: str | None = None
    due_at: str | None = None
    status: str | None = None
    priority_id: str | None = None

class Action(BaseModel):
    task: ActionTask

class CreatedAction(BaseModel):
    action_id: str

class User(BaseModel):
    id: str
    email: str
    firstname: str | None = None
    lastname: str | None = None
    active: bool = True

pytest tests/test_tools.py -v -k "test_inspection or test_template or test_action or test_user or test_created"

git add src/safetyculture_mcp/models/schemas.py tests/test_tools.py
git commit -m "feat: add pydantic v2 schemas for all domains"

import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from safetyculture_mcp.tools.inspections import list_inspections, get_inspection

def make_mock_client(status_code: int, json_body: dict) -> MagicMock:
    mock_resp = MagicMock()
    mock_resp.status_code = status_code
    mock_resp.json.return_value = json_body
    mock_resp.url = "http://mock"
    mock_client = AsyncMock()
    mock_client.get = AsyncMock(return_value=mock_resp)
    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
    mock_client.__aexit__ = AsyncMock(return_value=None)
    return mock_client

@pytest.mark.asyncio
async def test_list_inspections_returns_summaries():
    mock_client = make_mock_client(200, {
        "audits": [
            {"audit_id": "audit_1", "template_id": "tmpl_1", "modified_at": "2024-01-01"},
        ]
    })
    with patch("safetyculture_mcp.tools.inspections.httpx.AsyncClient", return_value=mock_client):
        result = await list_inspections()
    assert len(result) == 1
    assert result[0].audit_id == "audit_1"
    assert result[0].template_id == "tmpl_1"

@pytest.mark.asyncio
async def test_list_inspections_empty():
    mock_client = make_mock_client(200, {"audits": []})
    with patch("safetyculture_mcp.tools.inspections.httpx.AsyncClient", return_value=mock_client):
        result = await list_inspections()
    assert result == []

@pytest.mark.asyncio
async def test_list_inspections_401_raises_tool_error():
    from fastmcp.exceptions import ToolError
    mock_client = make_mock_client(401, {})
    with patch("safetyculture_mcp.tools.inspections.httpx.AsyncClient", return_value=mock_client):
        with pytest.raises(ToolError, match="Invalid or expired"):
            await list_inspections()

@pytest.mark.asyncio
async def test_get_inspection_returns_detail():
    mock_client = make_mock_client(200, {
        "inspection": {
            "id": "insp_1",
            "title": "Site Safety Check",
            "template_id": "tmpl_1",
            "is_marked_as_complete": True,
        }
    })
    with patch("safetyculture_mcp.tools.inspections.httpx.AsyncClient", return_value=mock_client):
        result = await get_inspection("insp_1")
    assert result.id == "insp_1"
    assert result.title == "Site Safety Check"
    assert result.is_marked_as_complete is True

@pytest.mark.asyncio
async def test_get_inspection_404_raises_tool_error():
    from fastmcp.exceptions import ToolError
    mock_client = make_mock_client(404, {})
    with patch("safetyculture_mcp.tools.inspections.httpx.AsyncClient", return_value=mock_client):
        with pytest.raises(ToolError, match="not found"):
            await get_inspection("bad_id")

[tool.pytest.ini_options]
asyncio_mode = "auto"

pytest tests/test_tools.py -v -k "test_list_inspections or test_get_inspection"

import httpx
from fastmcp import FastMCP
from ..client import BASE_URL, HEADERS, raise_for_status
from ..models.schemas import InspectionSummary, InspectionDetail

mcp = FastMCP(name="Inspections")

@mcp.tool(description="List inspections for the authenticated SafetyCulture account")
async def list_inspections(
    limit: int = 20,
    archived: bool = False,
    completed: bool | None = None,
) -> list[InspectionSummary]:
    params: dict = {"limit": limit, "archived": archived}
    if completed is not None:
        params["completed"] = completed
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{BASE_URL}/audits/search",
            headers=HEADERS,
            params=params,
        )
    raise_for_status(resp)
    return [InspectionSummary(**a) for a in resp.json().get("audits", [])]

@mcp.tool(description="Get full details for a single SafetyCulture inspection by ID")
async def get_inspection(inspection_id: str) -> InspectionDetail:
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{BASE_URL}/inspections/v1/inspections/{inspection_id}",
            headers=HEADERS,
        )
    raise_for_status(resp)
    return InspectionDetail(**resp.json()["inspection"])

pytest tests/test_tools.py -v -k "test_list_inspections or test_get_inspection"

git add src/safetyculture_mcp/tools/inspections.py tests/test_tools.py pyproject.toml
git commit -m "feat: add inspections tools (list_inspections, get_inspection)"

from safetyculture_mcp.tools.templates import list_templates

@pytest.mark.asyncio
async def test_list_templates_returns_list():
    mock_client = make_mock_client(200, {
        "templates": [
            {"template_id": "tmpl_1", "name": "Daily Checklist", "modified_at": "2024-01-01"},
        ]
    })
    with patch("safetyculture_mcp.tools.templates.httpx.AsyncClient", return_value=mock_client):
        result = await list_templates()
    assert len(result) == 1
    assert result[0].template_id == "tmpl_1"
    assert result[0].name == "Daily Checklist"

@pytest.mark.asyncio
async def test_list_templates_empty():
    mock_client = make_mock_client(200, {"templates": []})
    with patch("safetyculture_mcp.tools.templates.httpx.AsyncClient", return_value=mock_client):
        result = await list_templates()
    assert result == []

@pytest.mark.asyncio
async def test_list_templates_429_raises_tool_error():
    from fastmcp.exceptions import ToolError
    mock_client = make_mock_client(429, {})
    with patch("safetyculture_mcp.tools.templates.httpx.AsyncClient", return_value=mock_client):
        with pytest.raises(ToolError, match="rate limit"):
            await list_templates()

pytest tests/test_tools.py -v -k "test_list_templates"

import httpx
from fastmcp import FastMCP
from ..client import BASE_URL, HEADERS, raise_for_status
from ..models.schemas import Template

mcp = FastMCP(name="Templates")

@mcp.tool(description="List templates for the authenticated SafetyCulture account")
async def list_templates(
    limit: int = 20,
    archived: bool = False,
) -> list[Template]:
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{BASE_URL}/templates/search",
            headers=HEADERS,
            params={"limit": limit, "archived": archived},
        )
    raise_for_status(resp)
    return [Template(**t) for t in resp.json().get("templates", [])]

pytest tests/test_tools.py -v -k "test_list_templates"

git add src/safetyculture_mcp/tools/templates.py tests/test_tools.py
git commit -m "feat: add templates tools (list_templates)"

from safetyculture_mcp.tools.actions import list_actions, get_action, create_action

def make_mock_client_post(status_code: int, json_body: dict) -> MagicMock:
    mock_resp = MagicMock()
    mock_resp.status_code = status_code
    mock_resp.json.return_value = json_body
    mock_resp.url = "http://mock"
    mock_client = AsyncMock()
    mock_client.post = AsyncMock(return_value=mock_resp)
    mock_client.get = AsyncMock(return_value=mock_resp)
    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
    mock_client.__aexit__ = AsyncMock(return_value=None)
    return mock_client

@pytest.mark.asyncio
async def test_list_actions_returns_actions():
    mock_client = make_mock_client_post(200, {
        "actions": [
            {"task": {"task_id": "task_1", "title": "Fix hazard", "status": "open"}}
        ],
        "next_page_token": "",
    })
    with patch("safetyculture_mcp.tools.actions.httpx.AsyncClient", return_value=mock_client):
        result = await list_actions()
    assert len(result) == 1
    assert result[0].task.task_id == "task_1"
    assert result[0].task.title == "Fix hazard"

@pytest.mark.asyncio
async def test_list_actions_empty():
    mock_client = make_mock_client_post(200, {"actions": []})
    with patch("safetyculture_mcp.tools.actions.httpx.AsyncClient", return_value=mock_client):
        result = await list_actions()
    assert result == []

@pytest.mark.asyncio
async def test_get_action_returns_action():
    mock_resp = MagicMock()
    mock_resp.status_code = 200
    mock_resp.json.return_value = {
        "action": {"task": {"task_id": "task_1", "title": "Fix hazard", "status": "open"}},
        "read_only": False,
    }
    mock_resp.url = "http://mock"
    mock_client = AsyncMock()
    mock_client.get = AsyncMock(return_value=mock_resp)
    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
    mock_client.__aexit__ = AsyncMock(return_value=None)
    with patch("safetyculture_mcp.tools.actions.httpx.AsyncClient", return_value=mock_client):
        result = await get_action("task_1")
    assert result.task.task_id == "task_1"

@pytest.mark.asyncio
async def test_create_action_returns_id():
    mock_client = make_mock_client_post(200, {"action_id": "new_action_1"})
    with patch("safetyculture_mcp.tools.actions.httpx.AsyncClient", return_value=mock_client):
        result = await create_action(title="Fix guard rail", description="North entrance")
    assert result.action_id == "new_action_1"

@pytest.mark.asyncio
async def test_list_actions_500_raises_tool_error():
    from fastmcp.exceptions import ToolError
    mock_client = make_mock_client_post(500, {})
    with patch("safetyculture_mcp.tools.actions.httpx.AsyncClient", return_value=mock_client):
        with pytest.raises(ToolError, match="server error"):
            await list_actions()

pytest tests/test_tools.py -v -k "test_list_actions or test_get_action or test_create_action"

import httpx
from fastmcp import FastMCP
from ..client import BASE_URL, HEADERS, raise_for_status
from ..models.schemas import Action, CreatedAction

mcp = FastMCP(name="Actions")

@mcp.tool(description="List actions for the authenticated SafetyCulture account")
async def list_actions(page_size: int = 20) -> list[Action]:
    # NOTE: List actions uses POST, not GET
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{BASE_URL}/tasks/v1/actions/list",
            headers=HEADERS,
            json={"page_size": page_size},
        )
    raise_for_status(resp)
    return [Action(**a) for a in resp.json().get("actions", [])]

@mcp.tool(description="Get full details for a single SafetyCulture action by ID")
async def get_action(action_id: str) -> Action:
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{BASE_URL}/tasks/v1/actions/{action_id}",
            headers=HEADERS,
        )
    raise_for_status(resp)
    return Action(**resp.json()["action"])

@mcp.tool(description="Create a new action in the authenticated SafetyCulture account")
async def create_action(
    title: str,
    description: str = "",
    due_at: str | None = None,
) -> CreatedAction:
    body: dict = {"title": title, "description": description}
    if due_at is not None:
        body["due_at"] = due_at
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{BASE_URL}/tasks/v1/actions",
            headers=HEADERS,
            json=body,
        )
    raise_for_status(resp)
    return CreatedAction(**resp.json())

pytest tests/test_tools.py -v -k "test_list_actions or test_get_action or test_create_action"

git add src/safetyculture_mcp/tools/actions.py tests/test_tools.py
git commit -m "feat: add actions tools (list_actions, get_action, create_action)"

from safetyculture_mcp.tools.users import list_users, search_users_by_email

@pytest.mark.asyncio
async def test_list_users_returns_users():
    mock_resp = MagicMock()
    mock_resp.status_code = 200
    mock_resp.json.return_value = {
        "data": [
            {"id": "user_1", "email": "alice@example.com", "firstname": "Alice", "lastname": "Smith", "active": True},
        ],
        "metadata": {"next_page": "", "remaining_records": 0},
    }
    mock_resp.url = "http://mock"
    mock_client = AsyncMock()
    mock_client.get = AsyncMock(return_value=mock_resp)
    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
    mock_client.__aexit__ = AsyncMock(return_value=None)
    with patch("safetyculture_mcp.tools.users.httpx.AsyncClient", return_value=mock_client):
        result = await list_users()
    assert len(result) == 1
    assert result[0].id == "user_1"
    assert result[0].email == "alice@example.com"
    assert result[0].firstname == "Alice"

@pytest.mark.asyncio
async def test_list_users_empty():
    mock_resp = MagicMock()
    mock_resp.status_code = 200
    mock_resp.json.return_value = {"data": [], "metadata": {"next_page": "", "remaining_records": 0}}
    mock_resp.url = "http://mock"
    mock_client = AsyncMock()
    mock_client.get = AsyncMock(return_value=mock_resp)
    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
    mock_client.__aexit__ = AsyncMock(return_value=None)
    with patch("safetyculture_mcp.tools.users.httpx.AsyncClient", return_value=mock_client):
        result = await list_users()
    assert result == []

@pytest.mark.asyncio
async def test_search_users_by_email_returns_users():
    mock_resp = MagicMock()
    mock_resp.status_code = 200
    mock_resp.json.return_value = {
        "users": [
            {"id": "user_2", "email": "bob@example.com", "firstname": "Bob", "lastname": "Jones", "active": True},
        ]
    }
    mock_resp.url = "http://mock"
    mock_client = AsyncMock()
    mock_client.post = AsyncMock(return_value=mock_resp)
    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
    mock_client.__aexit__ = AsyncMock(return_value=None)
    with patch("safetyculture_mcp.tools.users.httpx.AsyncClient", return_value=mock_client):
        result = await search_users_by_email(["bob@example.com"])
    assert len(result) == 1
    assert result[0].email == "bob@example.com"

pytest tests/test_tools.py -v -k "test_list_users or test_search_users"

import httpx
from fastmcp import FastMCP
from ..client import BASE_URL, HEADERS, raise_for_status
from ..models.schemas import User

mcp = FastMCP(name="Users")

@mcp.tool(description="List all users in the SafetyCulture organisation")
async def list_users() -> list[User]:
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{BASE_URL}/feed/users",
            headers=HEADERS,
        )
    raise_for_status(resp)
    return [User(**u) for u in resp.json().get("data", [])]

@mcp.tool(description="Search for SafetyCulture users by a list of email addresses")
async def search_users_by_email(emails: list[str]) -> list[User]:
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{BASE_URL}/users/search",
            headers=HEADERS,
            json={"email": emails},
        )
    raise_for_status(resp)
    return [User(**u) for u in resp.json().get("users", [])]

pytest tests/test_tools.py -v -k "test_list_users or test_search_users"

git add src/safetyculture_mcp/tools/users.py tests/test_tools.py
git commit -m "feat: add users tools (list_users, search_users_by_email)"

@pytest.mark.asyncio
async def test_server_mcp_exposes_all_tools():
    from safetyculture_mcp.server import mcp
    from fastmcp import FastMCP
    assert isinstance(mcp, FastMCP)
    assert mcp.name == "SafetyCulture MCP"
    # Verify all sub-apps are mounted by calling list_tools via the MCP protocol
    tools = await mcp.get_tools()
    tool_names = {name for name in tools}
    expected = {
        "list_inspections", "get_inspection",
        "list_templates",
        "list_actions", "get_action", "create_action",
        "list_users", "search_users_by_email",
    }
    assert expected.issubset(tool_names), f"Missing tools: {expected - tool_names}"

pytest tests/test_tools.py -v -k "test_server_mcp"

from fastmcp import FastMCP
from .tools.inspections import mcp as inspections_mcp
from .tools.actions import mcp as actions_mcp
from .tools.templates import mcp as templates_mcp
from .tools.users import mcp as users_mcp

mcp = FastMCP(name="SafetyCulture MCP")

mcp.mount(inspections_mcp)
mcp.mount(actions_mcp)
mcp.mount(templates_mcp)
mcp.mount(users_mcp)

if __name__ == "__main__":
    mcp.run()

pytest tests/test_tools.py -v -k "test_server_mcp"

pytest tests/test_tools.py -v

fastmcp inspect src/safetyculture_mcp/server.py:mcp

git add src/safetyculture_mcp/server.py tests/test_tools.py
git commit -m "feat: compose all sub-apps into root server"

# SafetyCulture MCP Server

A FastMCP server that exposes SafetyCulture inspections, actions, templates, and users as MCP tools.

## Setup

### Prerequisites
- Python 3.12+
- A SafetyCulture API token ([get one here](https://app.safetyculture.com/account/api))

### Install

```bash
python3.12 -m venv .venv
.venv\Scripts\activate # Windows
# source .venv/bin/activate # macOS/Linux
pip install --upgrade pip
pip install -r requirements.txt
pip install -e .

cp .env.example .env
# Edit .env and set SAFETYCULTURE_API_TOKEN=your_token_here

python -c "
import asyncio, httpx
from safetyculture_mcp.client import BASE_URL, HEADERS
async def check():
    async with httpx.AsyncClient() as c:
        r = await c.get(f'{BASE_URL}/feed/v2/accounts/me', headers=HEADERS)
        print(r.status_code, r.json())
asyncio.run(check())
"

pip install -r requirements-dev.txt
pytest tests/ -v

fastmcp inspect src/safetyculture_mcp/server.py:mcp

{
  "mcpServers": {
    "safetyculture": {
      "command": "fastmcp",
      "args": ["run", "C:/path/to/safetyculture-mcp/src/safetyculture_mcp/server.py:mcp"],
      "env": {
        "SAFETYCULTURE_API_TOKEN": "your_token_here"
      }
    }
  }
}

- [ ] **Step 2: Commit**

```bash
git add README.md
git commit -m "docs: add README with setup, tools reference, and Prefect Horizon deploy steps"

pytest tests/ -v

fastmcp version

FastMCP version: 3.4.2
MCP version: 1.27.2
Python version: 3.12.x