orchestrata_ultimate.tools.mcp_resource_tools

Reference › orchestrata_ultimate › tools › mcp_resource_tools

This module provides the two MCP resource tools of SPEC-071, in MVP scope: ListMcpResourcesTool and ReadMcpResourceTool. Each contacts one MCP stdio server per call through mcp_stdio_client. They are BaseTool subclasses in the ToolCategory for MCP, registered by agents.tool_registry and exported from the package __init__. tests/tools/test_mcp_resource_tools.py covers them. The classes carry @traces for the REQ-071 series.

The caller supplies the server command and args and a timeout in the tool arguments. ListMcpResourcesTool.execute validates them, connects, lists resources, optionally filters by MIME type prefix, and applies offset and limit on the client side. It returns a JSON payload (ListResourcesOutput) with the resource list (MCPResourceInfo), a server status record (MCPServerStatus) and the total count. ReadMcpResourceTool.execute reads one URI and returns text content or base64 bytes (ReadResourceOutput), with client-side slicing by offset and limit through _slice_text and _slice_bytes. Neither tool raises. Bad arguments produce a validation error, and transport or protocol failures produce a failed ToolResult that carries the client's error code and message.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

MCPResourceInfo (dataclass)

Decorators: traces(spec='071', story='US-071-001', reqs=['REQ-071-007', 'REQ-071-008']), dataclass · Line 52

One resource entry (REQ-071-007/008) — the MVP subset of spec.md's MCPResource.

MCPServerStatus (dataclass)

Decorators: traces(spec='071', story='US-071-001', reqs=['REQ-071-012', 'REQ-071-014', 'REQ-071-015']), dataclass · Line 73

Status of the single server contacted for this call (REQ-071-012/014/015).

ListResourcesOutput (dataclass)

Decorators: traces(spec='071', story='US-071-001', reqs=['REQ-071-001', 'REQ-071-006']), dataclass · Line 93

Result payload for :class:ListMcpResourcesTool (JSON-encoded into ToolResult.output).

ReadResourceOutput (dataclass)

Decorators: traces(spec='071', story='US-071-002', reqs=['REQ-071-021', 'REQ-071-022', 'REQ-071-023']), dataclass · Line 110

Result payload for :class:ReadMcpResourceTool (JSON-encoded into ToolResult.output).

ListMcpResourcesTool

Bases: BaseTool · Decorators: traces(spec='071', story='US-071-001', reqs=['REQ-071-001', 'REQ-071-004', 'REQ-071-006', 'REQ-071-007', 'REQ-071-008',… · Line 203

List resources exposed by one MCP stdio server (SPEC-071, MVP scope).

ReadMcpResourceTool

Bases: BaseTool · Decorators: traces(spec='071', story='US-071-002', reqs=['REQ-071-019', 'REQ-071-020', 'REQ-071-021', 'REQ-071-022', 'REQ-071-023',… · Line 357

Read one resource's content from an MCP stdio server (SPEC-071, MVP scope).

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/tools/mcp_resource_tools.py

"""ListMcpResourcesTool / ReadMcpResourceTool — SPEC-071 MCP resource tools (MVP).

Enumerate and read resources exposed by an MCP server. spec.md's full design
(``ListMcpResourcesTool``/``ReadMcpResourceTool``/``McpAuthTool``) assumes
infrastructure that doesn't exist yet in this codebase — a server registry,
a connection pool, a credential vault, and a Redis-backed resource cache —
and names the official ``mcp`` Python SDK as the transport dependency, which
is **not declared** in ``pyproject.toml``. Per the MVP scope for this build:

- Only ``ListMcpResourcesTool``/``ReadMcpResourceTool`` are implemented here;
  ``McpAuthTool`` (FR-009/010/011 — credential configuration, connection
  testing, credential management) is out of scope and not built.
- There is no server registry: callers pass the server's stdio launch
  ``command``/``args`` directly as tool parameters, and each call talks to
  exactly one server (no cross-server aggregation).
- Transport is :class:`~orchestrata_ultimate.tools.mcp_stdio_client.McpStdioClient`,
  a ~200-line hand-written JSON-RPC-over-stdio client (see that module's
  docstring for the full "Spec fork" rationale) rather than the official SDK.
- There is no cache, no versioning/checksum layer, and no PAM++ governance
  predicate wiring (spec.md's ``may_access_mcp/3`` etc.) — every call fetches
  fresh content directly from the server with no access-control gate beyond
  whatever the process invoking this tool already enforces.

See ``specs/071-mcp-resource-tools/checklists/requirements.md`` for exactly
which FR-* bullets are implemented (checked) vs. deferred (unchecked) under
this MVP.
"""

from __future__ import annotations

import base64
import json
from dataclasses import asdict, dataclass, field
from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.tools.base import (
    BaseTool,
    ToolCategory,
    ToolMetadata,
    ToolPriority,
    ToolResult,
)
from orchestrata_ultimate.tools.mcp_stdio_client import McpClientError, McpStdioClient

# ── Result payloads (MVP subset of spec.md's Data Models) ────────────────


@traces(spec="071", story="US-071-001", reqs=["REQ-071-007", "REQ-071-008"])
@dataclass
class MCPResourceInfo:
    """One resource entry (REQ-071-007/008) — the MVP subset of spec.md's ``MCPResource``.

    ``created_at``/``modified_at``/``version``/``is_protected`` are not part
    of the base MCP ``resources/list`` response and are omitted here rather
    than faked (see module docstring "Spec fork").
    """

    uri: str
    name: str
    mime_type: str | None = None
    size: int | None = None
    description: str | None = None


@traces(
    spec="071",
    story="US-071-001",
    reqs=["REQ-071-012", "REQ-071-014", "REQ-071-015"],
)
@dataclass
class MCPServerStatus:
    """Status of the single server contacted for this call (REQ-071-012/014/015).

    MVP substitute for spec.md's ``MCPConnection`` — there is no persistent
    connection pool/server registry (see module docstring), so this only
    describes the one command+args pair the caller supplied for this call.
    """

    command: str
    args: tuple[str, ...]
    status: str
    server_name: str | None = None
    server_version: str | None = None
    resource_count: int = 0
    last_used: str | None = None
    last_error: str | None = None


@traces(spec="071", story="US-071-001", reqs=["REQ-071-001", "REQ-071-006"])
@dataclass
class ListResourcesOutput:
    """Result payload for :class:`ListMcpResourcesTool` (JSON-encoded into ``ToolResult.output``)."""

    success: bool
    resources: list[MCPResourceInfo] = field(default_factory=list)
    server: MCPServerStatus | None = None
    total_count: int = 0
    error_message: str | None = None
    error_code: str | None = None


@traces(
    spec="071",
    story="US-071-002",
    reqs=["REQ-071-021", "REQ-071-022", "REQ-071-023"],
)
@dataclass
class ReadResourceOutput:
    """Result payload for :class:`ReadMcpResourceTool` (JSON-encoded into ``ToolResult.output``)."""

    success: bool
    uri: str
    content: str | None = None
    bytes_content_b64: str | None = None
    mime_type: str | None = None
    size: int = 0
    error_message: str | None = None
    error_code: str | None = None


def _validation_error(tool_name: str, message: str) -> ToolResult:
    """A failed :class:`ToolResult` for a bad-arguments call (never reaches the server)."""
    payload = {
        "success": False,
        "error_message": message,
        "error_code": "INVALID_ARGUMENTS",
    }
    return ToolResult(tool_name, False, json.dumps(payload), message)


@traces(spec="071", story="FR-008", reqs=["REQ-071-033", "REQ-071-034"])
def _slice_text(text: str, offset: int, limit: int | None) -> str:
    """Client-side offset/limit slicing of already-fetched text content (REQ-071-020/033/034)."""
    start = max(offset, 0)
    end = start + limit if limit is not None else None
    return text[start:end]


@traces(spec="071", story="FR-008", reqs=["REQ-071-033", "REQ-071-034"])
def _slice_bytes(raw: bytes, offset: int, limit: int | None) -> bytes:
    """Client-side offset/limit slicing of already-fetched binary content (REQ-071-020/033/034)."""
    start = max(offset, 0)
    end = start + limit if limit is not None else None
    return raw[start:end]


# ── ListMcpResourcesTool ──────────────────────────────────────────────

_LIST_RESOURCES_PARAMETERS: dict[str, Any] = {
    "type": "object",
    "properties": {
        "command": {
            "type": "string",
            "description": (
                "Executable that launches the MCP server over stdio. MVP note: "
                "SPEC-071 assumes a pre-registered server registry; none exists "
                "yet in this codebase, so callers name the server's launch "
                "command directly (see module docstring 'Spec fork')."
            ),
        },
        "args": {
            "type": "array",
            "items": {"type": "string"},
            "description": "Arguments passed to the server command.",
        },
        "filter_type": {
            "type": "string",
            "description": "Only return resources whose MIME type starts with this prefix (e.g. 'text/').",
        },
        "limit": {
            "type": "integer",
            "description": "Maximum number of resources to return.",
        },
        "offset": {
            "type": "integer",
            "description": "Number of matching resources to skip before applying limit.",
        },
        "timeout": {
            "type": "number",
            "description": "Seconds to wait for the server to respond before failing (default 10.0).",
        },
    },
    "required": ["command"],
}


@traces(
    spec="071",
    story="US-071-001",
    reqs=[
        "REQ-071-001",
        "REQ-071-004",
        "REQ-071-006",
        "REQ-071-007",
        "REQ-071-008",
        "REQ-071-012",
        "REQ-071-014",
        "REQ-071-015",
    ],
)
class ListMcpResourcesTool(BaseTool):
    """List resources exposed by one MCP stdio server (SPEC-071, MVP scope).

    Spec fork (see module docstring): SPEC-071's ``ListMcpResourcesTool`` is
    written against a server *registry* that aggregates resources across many
    already-configured, already-authenticated servers. No such registry
    exists in this codebase, so this MVP takes the server's launch
    ``command``/``args`` as tool parameters and talks to exactly one server
    per call via :class:`~orchestrata_ultimate.tools.mcp_stdio_client.McpStdioClient`.
    """

    def __init__(self) -> None:
        self._metadata = ToolMetadata(
            name="list_mcp_resources",
            description=(
                "List resources exposed by an MCP server reached over stdio "
                "(pass its launch command/args); supports MIME-type filtering "
                "and offset/limit pagination."
            ),
            category=ToolCategory.MCP,
            parameters=_LIST_RESOURCES_PARAMETERS,
            priority=ToolPriority.P3,
            tags=("mcp", "resources", "discovery"),
        )

    @property
    def metadata(self) -> ToolMetadata:
        return self._metadata

    @traces(spec="071", story="US-071-001", reqs=["REQ-071-004", "REQ-071-006"])
    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        """List resources; never raises — transport/protocol failures become a failed ``ToolResult``."""
        command = arguments.get("command")
        if not isinstance(command, str) or not command.strip():
            return _validation_error(
                self.name, "'command' is required and must be a non-empty string"
            )

        args = arguments.get("args") or []
        if not isinstance(args, list) or not all(isinstance(a, str) for a in args):
            return _validation_error(self.name, "'args' must be a list of strings")

        filter_type = arguments.get("filter_type")
        offset = int(arguments.get("offset") or 0)
        raw_limit = arguments.get("limit")
        limit = int(raw_limit) if raw_limit is not None else None
        timeout = float(arguments.get("timeout") or 10.0)

        try:
            async with McpStdioClient(command, args, timeout=timeout) as client:
                entries = await client.list_resources()
                server_info = client.server_info
        except McpClientError as exc:
            output = ListResourcesOutput(
                success=False,
                server=MCPServerStatus(
                    command=command,
                    args=tuple(args),
                    status="error",
                    last_error=str(exc),
                ),
                error_message=str(exc),
                error_code=exc.error_code,
            )
            return ToolResult(self.name, False, json.dumps(asdict(output)), str(exc))

        resources = [
            MCPResourceInfo(
                uri=e.uri,
                name=e.name,
                mime_type=e.mime_type,
                size=e.size,
                description=e.description,
            )
            for e in entries
        ]
        if filter_type:
            resources = [
                r for r in resources if (r.mime_type or "").startswith(filter_type)
            ]

        total_count = len(resources)
        if limit is not None:
            resources = resources[offset : offset + limit]
        else:
            resources = resources[offset:]

        server = MCPServerStatus(
            command=command,
            args=tuple(args),
            status="connected",
            server_name=server_info.get("name"),
            server_version=server_info.get("version"),
            resource_count=total_count,
            last_used=datetime.now(UTC).isoformat(),
        )
        output = ListResourcesOutput(
            success=True, resources=resources, server=server, total_count=total_count
        )
        return ToolResult(self.name, True, json.dumps(asdict(output)))


# ── ReadMcpResourceTool ───────────────────────────────────────────────

_READ_RESOURCE_PARAMETERS: dict[str, Any] = {
    "type": "object",
    "properties": {
        "command": {
            "type": "string",
            "description": (
                "Executable that launches the MCP server over stdio (see "
                "list_mcp_resources for the same MVP server-registry fork note)."
            ),
        },
        "args": {
            "type": "array",
            "items": {"type": "string"},
            "description": "Arguments passed to the server command.",
        },
        "resource_uri": {
            "type": "string",
            "description": "URI of the resource to read, as returned by list_mcp_resources.",
        },
        "offset": {
            "type": "integer",
            "description": "Character (text) or byte (binary) offset to start reading from.",
        },
        "limit": {
            "type": "integer",
            "description": "Maximum number of characters/bytes to return.",
        },
        "timeout": {
            "type": "number",
            "description": "Seconds to wait for the server to respond before failing (default 10.0).",
        },
    },
    "required": ["command", "resource_uri"],
}


@traces(
    spec="071",
    story="US-071-002",
    reqs=[
        "REQ-071-019",
        "REQ-071-020",
        "REQ-071-021",
        "REQ-071-022",
        "REQ-071-023",
        "REQ-071-033",
        "REQ-071-034",
        "REQ-071-036",
    ],
)
class ReadMcpResourceTool(BaseTool):
    """Read one resource's content from an MCP stdio server (SPEC-071, MVP scope).

    Spec fork (see module docstring): no cache/versioning/checksum layer —
    every call fetches fresh content directly from the server. offset/limit
    are applied client-side to the fetched content (the base MCP
    ``resources/read`` response is not itself range-addressable over the
    wire), which is the honest MVP reading of FR-008's "partial reads".
    """

    def __init__(self) -> None:
        self._metadata = ToolMetadata(
            name="read_mcp_resource",
            description=(
                "Read a resource's content from an MCP server reached over "
                "stdio (pass its launch command/args and the resource URI); "
                "supports client-side offset/limit slicing."
            ),
            category=ToolCategory.MCP,
            parameters=_READ_RESOURCE_PARAMETERS,
            priority=ToolPriority.P3,
            tags=("mcp", "resources", "read"),
        )

    @property
    def metadata(self) -> ToolMetadata:
        return self._metadata

    @traces(
        spec="071",
        story="US-071-002",
        reqs=["REQ-071-033", "REQ-071-034", "REQ-071-036"],
    )
    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        """Read a resource; never raises — transport/protocol failures become a failed ``ToolResult``."""
        command = arguments.get("command")
        if not isinstance(command, str) or not command.strip():
            return _validation_error(
                self.name, "'command' is required and must be a non-empty string"
            )

        resource_uri = arguments.get("resource_uri")
        if not isinstance(resource_uri, str) or not resource_uri.strip():
            return _validation_error(
                self.name, "'resource_uri' is required and must be a non-empty string"
            )

        args = arguments.get("args") or []
        if not isinstance(args, list) or not all(isinstance(a, str) for a in args):
            return _validation_error(self.name, "'args' must be a list of strings")

        offset = int(arguments.get("offset") or 0)
        raw_limit = arguments.get("limit")
        limit = int(raw_limit) if raw_limit is not None else None
        timeout = float(arguments.get("timeout") or 10.0)

        try:
            async with McpStdioClient(command, args, timeout=timeout) as client:
                contents = await client.read_resource(resource_uri)
        except McpClientError as exc:
            output = ReadResourceOutput(
                success=False,
                uri=resource_uri,
                error_message=str(exc),
                error_code=exc.error_code,
            )
            return ToolResult(self.name, False, json.dumps(asdict(output)), str(exc))

        if not contents:
            message = f"resource returned no content: {resource_uri}"
            output = ReadResourceOutput(
                success=False,
                uri=resource_uri,
                error_message=message,
                error_code="RESOURCE_NOT_FOUND",
            )
            return ToolResult(self.name, False, json.dumps(asdict(output)), message)

        item = contents[0]
        if item.text is not None:
            sliced_text = _slice_text(item.text, offset, limit)
            output = ReadResourceOutput(
                success=True,
                uri=item.uri,
                content=sliced_text,
                mime_type=item.mime_type,
                size=len(sliced_text),
            )
        else:
            raw = base64.b64decode(item.blob) if item.blob else b""
            sliced_bytes = _slice_bytes(raw, offset, limit)
            output = ReadResourceOutput(
                success=True,
                uri=item.uri,
                bytes_content_b64=base64.b64encode(sliced_bytes).decode("ascii"),
                mime_type=item.mime_type,
                size=len(sliced_bytes),
            )
        return ToolResult(self.name, True, json.dumps(asdict(output)))