MCP Operations Server

An MCP (Model Context Protocol) server that turns a read-only operations and telemetry REST API into 45 agent tools across 12 domains: asset inventory, reference data, aggregated metrics, rankings, sessions, device protocol messages, active outages, incidence management, power panel telemetry, user permissions and platform settings. The interesting part is not the tool count but the shape of the problem: an MCP server that fronts a private, signature-authenticated HTTP API has to deal with credential rotation, a synchronous HTTP client inside an async protocol server, a tool surface large enough that descriptions become the actual interface, and the fact that nobody can run it without the backend. This repository addresses the last point with a fixture mode: OPS_API_MODE=mock swaps the signed HTTP client for a provider that serves synthetic JSON from files in the repository, so the full tool catalogue is runnable, testable and demonstrable with no backend and no credentials.

Architecture

                    tool call (JSON-RPC over stdio or streamable HTTP)
                                        |
+---------------------+                 v
|     MCP client      |      +--------------------------+
| (desktop app, IDE,  |----->|   MCP Operations Server  |
|  agent framework)   |      |        (run.py)          |
+---------------------+      +--------------------------+
                                        |
                       registers 45 tools across 12 domains
                                        |
     +----------------+-----------------+-----------------+----------------+
     |                |                 |                 |                |
  assets           masters         aggregations          ranks          sessions
  messages        downtime         incidences           panels          users
  settings   incidence_triggers
     |                |                 |                 |                |
     +----------------+--------+--------+-----------------+----------------+
                               |
                      ApiClient (protocol)
                               |
              OPS_API_MODE ----+----------------------------+
                               |                            |
                 mock          v                     live   v
        +------------------------------+   +-------------------------------+
        |        MockApiClient         |   |     SignedHttpApiClient       |
        | FixtureProvider + routes.json|   | requests + AWS SigV4 (AWS4Auth)|
        |  mcp_operations/mock/fixtures|   |   -> HTTPS operations API      |
        +------------------------------+   +-------------------------------+

Both clients implement the same ApiClient protocol (async get(path, params) -> dict), so the domains never know which one they are talking to.

Related MCP server: Datadog MCP Server

Features

• 45 read-only tools over 12 domains, each declared by a decorated async method whose signature becomes the JSON schema and whose docstring becomes the tool description.

• Mock mode: a route manifest (routes.json) maps request paths to JSON fixtures, with single-segment path templates, literal-before-template matching and working skip/limit pagination. Adding a mock route is one file plus one manifest entry, no Python change.

• Live mode: AWS SigV4 request signing through AWS4Auth, resolving credentials on every call so rotating role credentials do not start failing with 403 after an hour.

• Blocking HTTP offloaded with asyncio.to_thread, so a slow upstream call does not block the event loop serving other tool calls.

• Two transports from the same code: streamable HTTP for containers, stdio for desktop clients, plus a .mcpb desktop bundle.

• Configuration validated at start-up: an unknown mode or a live mode without a base URL fails immediately instead of failing mid tool call.

• 67 unit tests covering the signing path (never a real request), the registered tool surface, the mock provider and the configuration.

Prerequisites

• Python 3.10 or newer.

• Nothing else for mock mode.

• For live mode: network access to an operations API that accepts SigV4-signed requests, and AWS credentials (role, profile or static keys) allowed to invoke it.

• For the desktop bundle: node/npx on the build machine and uv on the machine that installs the extension.

Configuration

.env.example lists the same variables with placeholder values. Copy it to .env for local use; .env is ignored by git.

Installation

python -m venv .venv
source .venv/bin/activate
pip install -r requirements-dev.txt   # requirements.txt for a runtime-only install

Usage

Mock mode

Mock mode is the default, so no configuration is needed:

python run.py
# -> http://127.0.0.1:8000/mcp

The log line Mock provider ready: 32 routes confirms the fixture set loaded. Every response carries a _mock block naming the route, the fixture file and the parameters that were applied, which makes it obvious that the data is synthetic:

{
  "total": 3,
  "results": [{"device_id": "dev-0001", "name": "Device Alpha 01"}],
  "_mock": {
    "mode": "mock",
    "route": "/assets/device",
    "fixture": "assets_device_list.json",
    "path": "/assets/device",
    "path_params": {},
    "query_params": {"limit": 1}
  }
}

To extend or replace the data, add a JSON file under mcp_operations/mock/fixtures/, register it in routes.json, or point OPS_MOCK_FIXTURES_DIR at a directory of your own.

Live mode

export OPS_API_MODE=live
export OPS_API_BASE_URL="https://api.example.com/v1"
export OPS_API_REGION=eu-west-1
# credentials from a role, a profile, or AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY
python run.py

Connecting an MCP client

Over stdio, which is what most desktop clients and IDEs expect:

{
  "mcpServers": {
    "operations": {
      "command": "/absolute/path/to/mcp-operations-server/.venv/bin/python",
      "args": ["/absolute/path/to/mcp-operations-server/run.py"],
      "env": {
        "MCP_TRANSPORT": "stdio",
        "OPS_API_MODE": "mock"
      }
    }
  }
}

Over HTTP, for clients that connect by URL:

{
  "mcpServers": {
    "operations": {
      "url": "http://localhost:8000/mcp"
    }
  }
}

A quick check that the transport is alive:

curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"probe","version":"1"}}}'

Container

docker build -t mcp-operations-server .
docker run --rm -p 8000:8000 -e MCP_HOST=0.0.0.0 mcp-operations-server

The image defaults to mock mode, so it runs with no credentials.

Desktop bundle (.mcpb)

chmod +x mcpb/build.sh
./mcpb/build.sh
# -> mcpb/dist/mcp-operations-server.mcpb

Drag the resulting file onto the desktop client to install it. The bundle runs over stdio with the uv runtime, resolving mcpb/pyproject.toml on the user's machine, and defaults to mock mode so it works without credentials. Build details, caveats and the fallback path are in mcpb/README.md.

Tool reference

Project structure

mcp-operations-server/
├── run.py                        Entry point: transport selection and signal handling
├── test.sh                       Lint and unit tests
├── Dockerfile                    Container image, mock mode by default
├── requirements.txt              Runtime dependencies (pinned)
├── requirements-dev.txt          Test and lint dependencies
├── .env.example                  Every environment variable with placeholder values
├── mcp_operations/
│   ├── config.py                 Environment parsing and validation (Settings)
│   ├── server.py                 Server assembly, instructions, domain registration
│   ├── api_client.py             ApiClient protocol, SignedHttpApiClient, MockApiClient
│   ├── domains/
│   │   ├── base.py               BaseDomain and the @tool decorator
│   │   ├── assets.py             Inventory tools
│   │   ├── masters.py            Reference data tool
│   │   ├── aggregations.py       Metric and KPI tools
│   │   ├── ranks.py              Ranking tools
│   │   ├── sessions.py           Session tools
│   │   ├── messages.py           Protocol message tools
│   │   ├── downtime.py           Active outage tool
│   │   ├── incidences.py         Incidence tools
│   │   ├── incidence_triggers.py Trigger rule tools
│   │   ├── panels.py             Power panel telemetry tools
│   │   ├── users.py              Permission tool
│   │   └── settings.py           Uptime setting tool
│   └── mock/
│       ├── provider.py           Route matching, fixture loading, pagination
│       └── fixtures/
│           ├── routes.json       Path template -> fixture file manifest
│           └── *.json            32 synthetic response fixtures, one per route
├── mcpb/                         Desktop extension packaging
│   ├── manifest.json             Extension metadata and configuration form
│   ├── pyproject.toml            Dependencies resolved on the user's machine
│   ├── stdio_server.py           stdio entry point
│   ├── build.sh                  Bundle build script
│   └── README.md                 Build, install and access-control notes
└── tests/unit/
    ├── test_config.py            Environment parsing and client selection
    ├── test_tool_decorator.py    Decorator marking and registration
    ├── test_tool_registration.py Tool surface: names, descriptions, schemas, vocabulary
    ├── test_api_client_signing.py SigV4 signing, with the network blocked
    └── test_mock_provider.py     Route matching, pagination, fixture integrity

Adding a domain

1. Create mcp_operations/domains/your_domain.py with a class extending BaseDomain.

2. Decorate each tool method with @tool. The signature becomes the input schema and the docstring becomes the description the model reads, so both are the interface.

3. Add the class to DOMAINS in mcp_operations/server.py.

4. Add a route and a fixture under mcp_operations/mock/fixtures/ so the new tools work in mock mode.

from mcp_operations.domains.base import BaseDomain, tool


class MyDomain(BaseDomain):
    @tool
    async def my_tool(self, param: str) -> dict:
        """Description the MCP client shows to the model."""
        return await self._api_client.get("/my/endpoint", {"param": param})

    @tool(name="custom_name")
    async def another_tool(self) -> dict:
        """Any kwargs passed to @tool are forwarded to mcp.add_tool()."""
        return await self._api_client.get("/another/endpoint")

Testing

./test.sh                  # flake8 (E501 ignored) plus the unit tests
pytest tests/unit -q       # tests only

67 tests, all offline. What they cover:

• test_api_client_signing.py asserts the Authorization header structure, the credential scope (date, region, service), the content hash and date headers, session token propagation, that a different query string produces a different signature, and that credentials are re-resolved on every call. An autouse fixture patches the transport adapter so a real HTTP request fails the test instead of leaving the machine.

• test_tool_registration.py builds the real server in mock mode and asserts that every domain tool is registered, that names are unique and snake_case, that the surface is read-only, that every tool has a description and an object schema whose required parameters exist, and that no retired domain vocabulary appears in any agent-visible string.

• test_mock_provider.py covers literal-before-template route matching, trailing-slash normalisation, pagination edge cases, fixture immutability across calls, custom fixture directories, and the integrity of the bundled set (every route resolves to a file, every file is reachable from a route, every file is valid JSON).

• test_config.py covers defaults, region fallback, and the failures that must happen at start-up rather than mid call.

Limitations and notes

• Read-only. Every tool is a GET; nothing in this server mutates upstream state.

• The streamable HTTP transport has no authentication or authorization of its own. It binds 127.0.0.1 by default for that reason. Exposing it (MCP_HOST=0.0.0.0, a published container port) puts the whole tool catalogue in reach of anyone who can connect, so put an authenticating gateway or reverse proxy in front of it and scope the upstream credentials to read-only.

• Mock data is synthetic and static. Fixtures do not simulate an upstream error, a timeout or a 403, and filter parameters other than skip and limit are echoed rather than applied. Mock mode demonstrates the interface, not the backend's query semantics.

• The HTTP client is synchronous requests offloaded to a thread, not a native async client. That costs one worker thread per in-flight request, bounded by the default thread pool. api_client.py documents what a fully async httpx plus SigV4Auth alternative would involve and why it is not used here.

• The uv server type used by the desktop bundle is flagged experimental in the MCPB specification and may change.

• The upstream API this server was written against is private, which is exactly why mock mode exists. The endpoint paths, parameter names and response shapes in this repository describe a generic operations API and are the contract the fixtures implement.

• Date parameters are strings in YYYYMMDDHHmmSS form and are passed through untouched; the server does not validate or convert them.

License

MIT. See LICENSE.