# 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.

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

| Variable | Description | Default |
|---|---|---|
| `OPS_API_MODE` | Backend selection. `mock` serves the bundled fixtures, `live` signs real HTTP calls. | `mock` |
| `OPS_API_BASE_URL` | Base URL of the upstream operations API. Required when mode is `live`. | empty |
| `OPS_API_REGION` | AWS region used in the signature credential scope. Falls back to `AWS_REGION`. | `eu-west-1` |
| `OPS_API_SERVICE` | AWS service name used in the signature credential scope. | `execute-api` |
| `OPS_API_TIMEOUT` | Per-request HTTP timeout, in seconds (live mode). | `30` |
| `OPS_MOCK_FIXTURES_DIR` | Directory of an alternative fixture set. Must contain `routes.json`. | bundled fixtures |
| `OPS_SERVER_VERSION` | Version reported to the client in `serverInfo`. | `1.0.0` |
| `OPS_LOG_LEVEL` | Root log level (`DEBUG`, `INFO`, `WARNING`, `ERROR`). | `INFO` |
| `MCP_TRANSPORT` | `streamable-http` or `stdio`. | `streamable-http` |
| `MCP_HOST` | Bind address for the HTTP transport. | `127.0.0.1` |
| `MCP_PORT` | Bind port for the HTTP transport. | `8000` |
| `MCP_PATH` | Mount path of the HTTP transport. | `/mcp` |
| `MCP_STATELESS` | Serve each HTTP request with a fresh transport and accept an externally supplied `Mcp-Session-Id`. | `true` |
| `AWS_ACCESS_KEY_ID` | Standard AWS credential variable, live mode only. | unset |
| `AWS_SECRET_ACCESS_KEY` | Standard AWS credential variable, live mode only. | unset |
| `AWS_SESSION_TOKEN` | Standard AWS credential variable, live mode only, when credentials are temporary. | unset |
| `AWS_REGION` | Standard AWS region variable, used when `OPS_API_REGION` is unset. | unset |

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

## Installation

```bash
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:

```bash
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:

```json
{
  "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

```bash
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:

```json
{
  "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:

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

A quick check that the transport is alive:

```bash
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

```bash
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)

```bash
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](mcpb/README.md).

## Tool reference

| Domain | Tool | Purpose |
|---|---|---|
| assets | `list_devices` | Paginated device list with inventory and location filters |
| assets | `get_device_detail` | Full device record: hardware, channels, payment terminal, upstream metadata |
| assets | `list_sites` | Paginated site list |
| assets | `get_site_detail` | Site record with every asset installed at it |
| assets | `list_batteries` | Paginated battery list |
| assets | `get_battery_detail` | Full battery record |
| assets | `list_load_controllers` | Paginated load controller list |
| assets | `get_load_controller_detail` | Full load controller record and managed devices |
| assets | `list_power_panels` | Paginated power panel list |
| assets | `get_power_panel_detail` | Full panel record and switch-to-device mapping |
| assets | `list_energy_meters` | Paginated energy meter list |
| assets | `get_energy_meter_detail` | Full energy meter record |
| masters | `get_master_data` | Distinct values of a reference field, for building and validating filters |
| aggregations | `get_instant_power` | Real-time aggregated power draw (W) |
| aggregations | `get_energy_consumed` | Accumulated energy (Wh) by period |
| aggregations | `get_availability_count` | Device count by availability bucket |
| aggregations | `get_uptime` | Uptime or downtime percentage per device, fleet or site |
| aggregations | `get_sessions_count` | Count of completed sessions by period |
| aggregations | `get_usage_time` | Occupied, available and faulted time split |
| aggregations | `get_installed_power` | Installed power capacity, as a count or a sum of Watts |
| aggregations | `get_registrations` | Device registrations over time |
| aggregations | `get_incidences_count` | Incidence count over time |
| aggregations | `get_panel_energy_consumed` | Energy measured at the power panels of a site |
| aggregations | `get_meter_energy_consumed` | Energy measured at the energy meters of a site |
| ranks | `get_sessions_rank` | Ranking by number of sessions, grouped by any dimension |
| ranks | `get_incidences_rank` | Ranking of devices by incidence count |
| ranks | `get_energy_rank` | Ranking by energy delivered |
| ranks | `get_uptime_rank` | Ranking of devices by average uptime |
| ranks | `get_peak_power_rank` | Ranking of devices by peak power in one session |
| ranks | `get_peak_energy_rank` | Ranking of devices by peak energy in one session |
| sessions | `list_sessions` | Individual session records with energy, peak power and levels |
| sessions | `get_session_curve` | Power and energy time series of one session |
| messages | `list_device_messages` | Protocol message history of a device for a period |
| messages | `get_raw_device_message` | Raw frame of a single message |
| downtime | `get_downtime_outages` | Devices currently down, with outage start and days elapsed |
| incidences | `list_incidences` | Incidence list with status, severity and classification filters |
| incidences | `get_incidence_detail` | Full incidence record with milestones and comments |
| incidences | `get_incidence_kpi` | One of seven incidence management KPIs |
| panels | `get_power_panel_telemetry` | Latest electrical and energy readings of a panel |
| panels | `get_power_panel_status` | Real-time panel alarms and per-switch status |
| users | `get_user_manufacturers` | Manufacturer scope granted to a user |
| settings | `get_uptime_settings` | Devices included in or excluded from the uptime calculation |
| incidence_triggers | `list_incidence_triggers` | Rules that raise incidences from device messages |
| incidence_triggers | `get_incidence_trigger` | One trigger rule in full |
| incidence_triggers | `get_incidence_trigger_asset_types` | Asset types, fields and operators available to rules |

## 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.

```python
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

```bash
./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](LICENSE).