From 5b0528ac5efea335268f99b05aed5b820347aeb7 Mon Sep 17 00:00:00 2001 From: Marc Christensen Date: Thu, 17 Sep 2026 22:07:06 -0600 Subject: [PATCH] Add compact 'summary' output format to list tools MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit list_containers, list_images and list_volumes returned the full inspect-level shape for every object, which is unusable at fleet scale: * 'state' embeds State.Health.Log — Docker keeps the last 5 healthcheck outputs and never bounds their size. A healthcheck that curls a web page (or prints a JSON status dump) adds tens of KB per container, per listing. * the nested 'image' object repeats repo tags, digests and OCI labels for every container row. * 'mounts' carries full mount definitions (source paths, drivers, propagation) for every row. Observed on a 22-container host: a single list_containers call returned ~103 KB, ~90% of it health-check log output. Add a format parameter ('summary' | 'full', default 'summary') to the three list tools. Summary returns a docker-ps-style row: name, short_id, status, state.{Status,Running,Restarting,Dead,ExitCode,StartedAt}, image (string ref, no nested object), mounts trimmed to name+type, health.{Status,FailingStreak} (no Log) plus a slim image/volume row (no labels/digests). Full keeps the existing inspect shape for debugging, and caps every Health.Log output entry at 500 characters in both modes, so even a full listing is bounded regardless of what a healthcheck prints. Single-object tools (run_container, stop_container, ...) are unchanged: docker_to_dict defaults to format='full'. Example output (real 22-container host, before -> after): list_containers 102,965 chars -> 11,048 chars (89% smaller) webapp-a row 24,013 chars -> 326 chars webapp-b row 19,629 chars -> 323 chars app-c row 11,201 chars -> 424 chars Verified against a live Docker daemon over ssh: 40 non-empty health-check outputs in full mode, 0 exceeding the cap. --- src/mcp_server_docker/output_schemas.py | 256 ++++++++++++---- src/mcp_server_docker/server.py | 24 +- tests/test_output_schemas.py | 386 ++++++++++++++++++++++++ tests/test_server.py | 6 +- 4 files changed, 608 insertions(+), 64 deletions(-) create mode 100644 tests/test_output_schemas.py diff --git a/src/mcp_server_docker/output_schemas.py b/src/mcp_server_docker/output_schemas.py index 48bf0d4..8d0a7f7 100644 --- a/src/mcp_server_docker/output_schemas.py +++ b/src/mcp_server_docker/output_schemas.py @@ -1,77 +1,215 @@ -from typing import Any +from typing import Any, Literal from docker.models.containers import Container from docker.models.images import Image from docker.models.networks import Network from docker.models.volumes import Volume +# Docker retains up to 5 healthcheck log entries and never bounds their size. +# A healthcheck that `curl`s a web page (or dumps a JSON status payload) +# therefore balloons every container listing by tens of KB each. Cap each +# entry's output so a `list_containers` call stays small no matter what the +# healthcheck prints. +HEALTH_LOG_ENTRY_CAP = 500 + +# The subset of a container's inspect State that a listing actually needs. +_CONTAINER_STATE_KEYS = ( + "Status", + "Running", + "Restarting", + "Dead", + "ExitCode", + "StartedAt", +) + + +def _cap_text(value: Any, limit: int = HEALTH_LOG_ENTRY_CAP) -> Any: + """Truncate a string to ``limit`` characters, tagging the cut. + + Non-string values (or short strings) are returned unchanged. + """ + if isinstance(value, str) and len(value) > limit: + return value[:limit] + "… [truncated]" + return value + + +def _trim_health_log(state: dict[str, Any] | None) -> dict[str, Any] | None: + """Return ``State`` with each ``Health.Log[*].Output`` entry capped. + + Returns the original object untouched when there is no health log, so + non-health-checked containers pay nothing for this. + """ + if not isinstance(state, dict): + return state + health = state.get("Health") + if not isinstance(health, dict) or not isinstance(health.get("Log"), list): + return state + log = [ + {**entry, "Output": _cap_text(entry.get("Output"))} + if isinstance(entry, dict) + else entry + for entry in health["Log"] + ] + return {**state, "Health": {**health, "Log": log}} + + +def _container_image_ref(obj: Container) -> str | None: + """Cheap image reference for a listing. + + Prefers ``Config.Image`` (no extra daemon round-trip). Falls back to the + first repo tag of the linked image when ``Config.Image`` is empty. + """ + config: dict[str, Any] = obj.attrs.get("Config") or {} + ref = config.get("Image") + if ref: + return ref + image = obj.image + if image is not None: + tags = getattr(image, "tags", None) or [] + if tags: + return tags[0] + return None + + +def _container_summary(obj: Container) -> dict[str, Any]: + """Compact per-container row: the ``docker ps`` columns plus health.""" + state: dict[str, Any] = obj.attrs.get("State") or {} + result: dict[str, Any] = { + "name": obj.name, + "short_id": obj.short_id, + "status": obj.status, + "state": {key: state.get(key) for key in _CONTAINER_STATE_KEYS}, + "image": _container_image_ref(obj), + "mounts": [ + {"name": mount.get("Name"), "type": mount.get("Type")} + for mount in (obj.attrs.get("Mounts") or []) + ], + } + health = state.get("Health") + if isinstance(health, dict): + result["health"] = { + "Status": health.get("Status"), + "FailingStreak": health.get("FailingStreak"), + } + return result + + +def _container_full(obj: Container) -> dict[str, Any]: + """Inspect-level container shape (pre-existing keys), health log capped.""" + config: dict[str, Any] = obj.attrs.get("Config") or {} + return { + "id": obj.id, + "name": obj.name, + "short_id": obj.short_id, + "image": docker_to_dict(obj.image) if obj.image else None, + "status": obj.status, + "labels": config.get("Labels", {}), + "ports": obj.ports, + "created": obj.attrs.get("Created"), + "state": _trim_health_log(obj.attrs.get("State")), + "restart_count": obj.attrs.get("RestartCount"), + "networks": list( + obj.attrs.get("NetworkSettings", {}).get("Networks", {}).keys() + ), + "mounts": obj.attrs.get("Mounts"), + "config": { + "hostname": config.get("Hostname"), + "user": config.get("User"), + "image": config.get("Image"), + }, + } + + +def _image_summary(obj: Image) -> dict[str, Any]: + return { + "short_id": obj.short_id, + "tags": obj.tags, + "size": obj.attrs.get("Size"), + "created": obj.attrs.get("Created"), + } + + +def _image_full(obj: Image) -> dict[str, Any]: + img_config: dict[str, Any] = obj.attrs.get("Config") or {} + return { + "id": obj.id, + "tags": obj.tags, + "short_id": obj.short_id, + "labels": img_config.get("Labels", {}), + "repo_tags": obj.attrs.get("RepoTags"), + "repo_digests": obj.attrs.get("RepoDigests"), + "created": obj.attrs.get("Created"), + "size": obj.attrs.get("Size"), + } + + +def _volume_summary(obj: Volume) -> dict[str, Any]: + return { + "name": obj.name, + "short_id": obj.short_id, + "driver": obj.attrs.get("Driver"), + "mountpoint": obj.attrs.get("Mountpoint"), + } + + +def _volume_full(obj: Volume) -> dict[str, Any]: + return { + "id": obj.id, + "name": obj.name, + "short_id": obj.short_id, + "labels": obj.attrs.get("Labels", {}), + "mountpoint": obj.attrs.get("Mountpoint"), + "created": obj.attrs.get("CreatedAt"), + "driver": obj.attrs.get("Driver"), + "scope": obj.attrs.get("Scope"), + } + + +def _network(obj: Network) -> dict[str, Any]: + return { + "id": obj.id, + "name": obj.name, + "short_id": obj.short_id, + "driver": obj.attrs.get("Driver"), + "scope": obj.attrs.get("Scope"), + "created": obj.attrs.get("CreatedAt"), + "labels": obj.attrs.get("Labels"), + } + def docker_to_dict( - obj: Image | Container | Volume | Network, overrides: dict[str, Any] | None = None + obj: Image | Container | Volume | Network, + overrides: dict[str, Any] | None = None, + format: Literal["summary", "full"] = "full", ) -> dict[str, Any]: - result = None + """Serialize a docker-py model to a plain dict. - if isinstance(obj, Image): - img_config: dict[str, Any] = obj.attrs.get("Config") or {} - - result = { - "id": obj.id, - "tags": obj.tags, - "short_id": obj.short_id, - "labels": img_config.get("Labels", {}), - "repo_tags": obj.attrs.get("RepoTags"), - "repo_digests": obj.attrs.get("RepoDigests"), - "created": obj.attrs.get("Created"), - "size": obj.attrs.get("Size"), - } + ``format`` selects the verbosity: + + * ``"summary"`` — a compact row tuned for listing (drops ``Health.Log``, + the nested image object, labels and digests). Use this for the + ``list_*`` tools so a listing of N containers stays small. + * ``"full"`` — the inspect-level shape used by the single-object tools + (``run_container``, ``stop_container``, …). Health log entries are + still capped to :data:`HEALTH_LOG_ENTRY_CAP` characters. + + ``format`` has no effect on networks (they are already lean). + """ + result = None if isinstance(obj, Container): - config: dict[str, Any] = obj.attrs.get("Config") or {} - - result = { - "id": obj.id, - "name": obj.name, - "short_id": obj.short_id, - "image": docker_to_dict(obj.image) if obj.image else None, - "status": obj.status, - "labels": config.get("Labels", {}), - "ports": obj.ports, - "created": obj.attrs.get("Created"), - "state": obj.attrs.get("State"), - "restart_count": obj.attrs.get("RestartCount"), - "networks": list( - obj.attrs.get("NetworkSettings", {}).get("Networks", {}).keys() - ), - "mounts": obj.attrs.get("Mounts"), - "config": { - "hostname": config.get("Hostname"), - "user": config.get("User"), - "image": config.get("Image"), - }, - } + result = ( + _container_summary(obj) if format == "summary" else _container_full(obj) + ) - if isinstance(obj, Network): - result = { - "id": obj.id, - "name": obj.name, - "short_id": obj.short_id, - "driver": obj.attrs.get("Driver"), - "scope": obj.attrs.get("Scope"), - "created": obj.attrs.get("CreatedAt"), - "labels": obj.attrs.get("Labels"), - } + if isinstance(obj, Image): + result = _image_summary(obj) if format == "summary" else _image_full(obj) if isinstance(obj, Volume): - result = { - "id": obj.id, - "name": obj.name, - "short_id": obj.short_id, - "labels": obj.attrs.get("Labels", {}), - "mountpoint": obj.attrs.get("Mountpoint"), - "created": obj.attrs.get("CreatedAt"), - "driver": obj.attrs.get("Driver"), - "scope": obj.attrs.get("Scope"), - } + result = _volume_summary(obj) if format == "summary" else _volume_full(obj) + + if isinstance(obj, Network): + result = _network(obj) if result is None: raise ValueError(f"Unsupported object type: {type(obj)}") diff --git a/src/mcp_server_docker/server.py b/src/mcp_server_docker/server.py index 59e835a..04249ba 100644 --- a/src/mcp_server_docker/server.py +++ b/src/mcp_server_docker/server.py @@ -66,6 +66,16 @@ class ListNetworksFilter(BaseModel): dict[str, str] | list[str] | None, Field(description="Container labels") ] AutoRemove = Annotated[bool, Field(description="Automatically remove the container")] +Format = Annotated[ + Literal["summary", "full"], + Field( + description=( + "Output verbosity. `summary` returns a compact `docker ps`-style row " + "(no health-check logs, labels or image digests); `full` returns the " + "inspect-level object. Defaults to `summary` for listings." + ) + ), +] def _client(ctx: Context[AppContext]) -> docker.DockerClient: @@ -239,9 +249,10 @@ def list_containers( filters: Annotated[ ListContainersFilters | None, Field(description="Filter containers") ] = None, + format: Format = "summary", ) -> list[dict[str, Any]]: return [ - docker_to_dict(container) + docker_to_dict(container, format=format) for container in _client(ctx).containers.list( all=all, filters=filters.model_dump() if filters else None ) @@ -491,9 +502,10 @@ def list_images( filters: Annotated[ ListImagesFilters | None, Field(description="Filter images") ] = None, + format: Format = "summary", ) -> list[dict[str, Any]]: return [ - docker_to_dict(image) + docker_to_dict(image, format=format) for image in _client(ctx).images.list( name=name, all=all, filters=filters.model_dump() if filters else None ) @@ -623,8 +635,12 @@ def remove_network( read_only_hint=True, idempotent_hint=True, open_world_hint=False ), ) -def list_volumes(ctx: Context[AppContext]) -> list[dict[str, Any]]: - return [docker_to_dict(volume) for volume in _client(ctx).volumes.list()] +def list_volumes( + ctx: Context[AppContext], format: Format = "summary" +) -> list[dict[str, Any]]: + return [ + docker_to_dict(volume, format=format) for volume in _client(ctx).volumes.list() + ] @app.tool( diff --git a/tests/test_output_schemas.py b/tests/test_output_schemas.py new file mode 100644 index 0000000..4c6f98f --- /dev/null +++ b/tests/test_output_schemas.py @@ -0,0 +1,386 @@ +"""Direct tests of the real ``docker_to_dict`` trimming logic. + +The existing ``test_server.py`` suite monkeypatches ``docker_to_dict`` with a +stub (so it only checks tool argument semantics). This file exercises the real +serialization — the summary shape, the full shape, the health-log cap, and the +image-ref fallback — with faithful docker-py models, plus one end-to-end MCP +call that proves ``list_containers`` defaults to the compact shape. +""" + +import json + +import pytest +from docker.models.containers import Container +from docker.models.images import Image +from docker.models.networks import Network +from docker.models.volumes import Volume +from mcp.client import Client + +import mcp_server_docker.server as server_module +from mcp_server_docker.output_schemas import ( + HEALTH_LOG_ENTRY_CAP, + _trim_health_log, + docker_to_dict, +) +from mcp_server_docker.server import app + +TRUNCATION_MARK = "\u2026 [truncated]" +BIG_OUTPUT = "" + "A" * 2000 + " use first repo tag of the linked image + attrs = _container_attrs(config_image="") + c = _Container(attrs, image=_image_obj(tags=("myrepo/app:latest", "myrepo/app:v1"))) + s = docker_to_dict(c, format="summary") + assert s["image"] == "myrepo/app:latest" + + +# --- full shape ------------------------------------------------------------- + + +def test_container_full_keeps_inspect_shape_and_caps_log(): + c = _rich_container() + f = docker_to_dict(c, format="full") + assert f["id"] == c.id + assert f["config"]["image"] == "nginx:latest" + assert "Health" in f["state"] + log = f["state"]["Health"]["Log"] + assert len(log) == 5 + for entry in log: + assert entry["Output"].endswith(TRUNCATION_MARK) + assert len(entry["Output"]) <= HEALTH_LOG_ENTRY_CAP + len(TRUNCATION_MARK) + # nested image present in full mode (sha256 short_id is 19 chars) + assert f["image"]["short_id"] == "sha256:9f1e2b3c4d5e" + # full is strictly larger than summary + assert len(json.dumps(f)) > len(json.dumps(docker_to_dict(c, format="summary"))) + + +def test_container_full_health_log_capped_even_when_shorter(): + # A short entry is returned unchanged (no marker). + health = { + "Status": "unhealthy", + "FailingStreak": 2, + "Log": [{"Start": "s", "End": "e", "ExitCode": 1, "Output": "ok\n"}], + } + c = _Container(_container_attrs(health=health), image=_image_obj()) + f = docker_to_dict(c, format="full") + assert f["state"]["Health"]["Log"][0]["Output"] == "ok\n" + + +# --- health-log trimming helper -------------------------------------------- + + +def test_trim_health_log_caps_each_entry(): + state = { + "Status": "running", + "Health": {"Status": "healthy", "Log": [{"Output": "y" * 3000}]}, + } + trimmed = _trim_health_log(state) + out = trimmed["Health"]["Log"][0]["Output"] + assert len(out) == HEALTH_LOG_ENTRY_CAP + len(TRUNCATION_MARK) + assert out.startswith("y" * HEALTH_LOG_ENTRY_CAP) + + +def test_trim_health_log_returns_nonhealth_state_untouched(): + state = {"Status": "exited"} + assert _trim_health_log(state) is state + assert _trim_health_log(None) is None + + +# --- images / volumes / networks ------------------------------------------- + + +def test_image_summary_and_full(): + img = _image_obj() + s = docker_to_dict(img, format="summary") + assert set(s.keys()) == {"short_id", "tags", "size", "created"} + assert s["tags"] == ["nginx:latest"] + f = docker_to_dict(img, format="full") + assert f["repo_digests"] == ["nginx@sha256:abcdef0123456789"] + assert "labels" in f + + +def test_volume_summary_and_full(): + vol = Volume( + attrs={ + "Id": "volid0000000000", + "Name": "app_data", + "Driver": "local", + "Mountpoint": "/var/lib/docker/volumes/app_data/_data", + "Labels": {"a": "b"}, + "CreatedAt": "2026-01-01T00:00:00Z", + "Scope": "local", + } + ) + s = docker_to_dict(vol, format="summary") + assert set(s.keys()) == {"name", "short_id", "driver", "mountpoint"} + f = docker_to_dict(vol, format="full") + assert f["labels"] == {"a": "b"} + assert f["scope"] == "local" + + +def test_network_ignores_format(): + net = Network( + attrs={ + "Id": "net0000000000", + "Name": "bridge", + "Driver": "bridge", + "Scope": "local", + "CreatedAt": "2026-01-01T00:00:00Z", + "Labels": None, + } + ) + assert docker_to_dict(net, format="summary") == docker_to_dict(net, format="full") + + +# --- overrides ------------------------------------------------------------- + + +def test_overrides_still_applied(): + s = docker_to_dict( + _rich_container(), format="summary", overrides={"status": "removed"} + ) + assert s["status"] == "removed" + + +# --- end-to-end through the MCP tool --------------------------------------- + + +class _FakeCollection: + def __init__(self, items): + self.items = items + + def list(self, **kwargs): + return self.items + + +class _FakeClient: + def __init__(self, containers): + self.containers = _FakeCollection(containers) + self.images = _FakeCollection([]) + self.networks = _FakeCollection([]) + self.volumes = _FakeCollection([]) + self.closed = False + + def close(self): + self.closed = True + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +@pytest.mark.anyio +async def test_list_containers_default_is_summary(monkeypatch): + fake = _FakeClient([_rich_container()]) + monkeypatch.setattr(server_module.docker, "from_env", lambda: fake) + + async with Client(app, raise_exceptions=True) as client: + default = await client.call_tool("list_containers", {}) + full = await client.call_tool("list_containers", {"format": "full"}) + + assert fake.closed + + rows = default.structured_content.get("result") + assert isinstance(rows, list) and len(rows) == 1 + row = rows[0] + # summary: image is a string, no config key, no Health.Log + assert isinstance(row["image"], str) + assert "config" not in row + assert "Log" not in row["health"] + assert row["health"]["Status"] == "healthy" + + full_rows = full.structured_content.get("result") + full_row = full_rows[0] + # full: inspect-level shape with capped health log + assert full_row["config"]["image"] == "nginx:latest" + assert "Health" in full_row["state"] + assert full_row["state"]["Health"]["Log"][0]["Output"].endswith(TRUNCATION_MARK) + + # the compact default is a large fraction smaller than the full output + assert len(json.dumps(full_rows)) > len(json.dumps(rows)) * 3 + + +@pytest.mark.anyio +async def test_single_object_tools_default_to_full(monkeypatch): + """run/stop/start keep returning the full object by default.""" + created: dict = {} + + class _RunCollection: + def run(self, **kwargs): + created.update(kwargs) + return _rich_container() + + class _Client: + def __init__(self): + self.containers = _RunCollection() + self.images = _FakeCollection([]) + self.networks = _FakeCollection([]) + self.volumes = _FakeCollection([]) + self.closed = False + + def close(self): + self.closed = True + + fake = _Client() + monkeypatch.setattr(server_module.docker, "from_env", lambda: fake) + + async with Client(app, raise_exceptions=True) as client: + res = await client.call_tool("run_container", {"image": "nginx"}) + + row = res.structured_content + assert row["config"]["image"] == "nginx:latest" + assert "Health" in row["state"] diff --git a/tests/test_server.py b/tests/test_server.py index a6c64d0..f36f67a 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -105,7 +105,11 @@ def docker_client(monkeypatch): monkeypatch.setattr( server_module, "docker_to_dict", - lambda _obj, overrides=None: {"id": "object-id", **(overrides or {})}, + lambda _obj, overrides=None, **_kwargs: { + "id": "object-id", + **_kwargs, + **(overrides or {}), + }, ) monkeypatch.setattr(server_module.docker, "from_env", lambda: client) yield client