Part 6: Capabilities

"""Module 8: what our tools look like on the wire as MCP (protocol version 2026-07-28).

This is an in-process teaching shim, not a compliant MCP server: it has no
transport, no authorization, and only three methods. It shows the message shapes
so you can read real MCP traffic and see how any MCP server's tools become the
same tool specs our agent loop already uses. For real servers use an official SDK.
"""
from __future__ import annotations

import json
from pathlib import Path

from m08_agent import BillingStore, Tool, make_tools

PROTOCOL_VERSION = "2026-07-28"
READ_ONLY = {"search_kb", "read_article", "get_account", "get_invoice"}


def to_mcp_tool(tool: Tool) -> dict:
    """An MCP Tool definition. Annotations are hints for the host, and hosts must treat them as untrusted."""
    return {"name": tool.name, "description": tool.description, "inputSchema": tool.parameters,
            "annotations": {"readOnlyHint": tool.name in READ_ONLY,
                            "destructiveHint": tool.irreversible,
                            "idempotentHint": tool.name in READ_ONLY or tool.name == "issue_refund",
                            "openWorldHint": False}}


class McpShim:
    """Answers server/discover, tools/list, and tools/call from our tool registry."""

    def __init__(self, tools: dict[str, Tool]) -> None:
        self.tools = tools

    def handle(self, request: dict) -> dict:
        rid, method, params = request["id"], request["method"], request.get("params", {})
        if method == "server/discover":
            result = {"supportedVersions": [PROTOCOL_VERSION], "capabilities": {"tools": {"listChanged": False}},
                      "instructions": "Brightlane support tools. issue_refund needs human approval in the host."}
        elif method == "tools/list":
            result = {"tools": [to_mcp_tool(t) for t in self.tools.values()], "ttlMs": 300000, "cacheScope": "private"}
        elif method == "tools/call":
            tool = self.tools.get(params.get("name"))
            if tool is None:  # protocol error: the model cannot fix an unknown tool name by retrying arguments
                return {"jsonrpc": "2.0", "id": rid, "error": {"code": -32602, "message": f"Unknown tool: {params.get('name')}"}}
            try:
                value = tool.fn(**params.get("arguments", {}))
                result = {"content": [{"type": "text", "text": json.dumps(value)}], "structuredContent": value, "isError": False}
            except Exception as exc:  # tool execution error: returned to the model so it can self-correct
                result = {"content": [{"type": "text", "text": f"{type(exc).__name__}: {exc}"}], "isError": True}
        else:
            return {"jsonrpc": "2.0", "id": rid, "error": {"code": -32601, "message": f"Method not found: {method}"}}
        return {"jsonrpc": "2.0", "id": rid, "result": {"resultType": "complete", **result}}


def mcp_to_chat_spec(mcp_tool: dict, server_prefix: str) -> dict:
    """Turn an MCP tool into a Chat Completions tool spec, prefixed to avoid clashes between servers."""
    return {"type": "function", "function": {"name": f"{server_prefix}__{mcp_tool['name']}",
                                             "description": mcp_tool.get("description", ""),
                                             "parameters": mcp_tool["inputSchema"]}}


if __name__ == "__main__":
    Path("runs/m08").mkdir(parents=True, exist_ok=True)
    server = McpShim(make_tools(BillingStore("runs/m08/mcp_billing.json"), flaky_failures=0))
    meta = {"io.modelcontextprotocol/protocolVersion": PROTOCOL_VERSION,
            "io.modelcontextprotocol/clientInfo": {"name": "brightlane-agent", "version": "0.8"}}
    print(json.dumps(server.handle({"jsonrpc": "2.0", "id": 1, "method": "server/discover", "params": {"_meta": meta}})))
    listed = server.handle({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {"_meta": meta}})
    for t in listed["result"]["tools"]:
        hints = t["annotations"]
        print(f"  {t['name']:18} readOnly={hints['readOnlyHint']!s:5} destructive={hints['destructiveHint']}")
    ok = server.handle({"jsonrpc": "2.0", "id": 3, "method": "tools/call",
                        "params": {"_meta": meta, "name": "search_kb", "arguments": {"query": "duplicate charge refund"}}})
    print(json.dumps(ok)[:220])
    bad = server.handle({"jsonrpc": "2.0", "id": 4, "method": "tools/call",
                         "params": {"_meta": meta, "name": "get_invoice", "arguments": {"invoice_id": "INV-2026-999999"}}})
    print(json.dumps(bad))
    unknown = server.handle({"jsonrpc": "2.0", "id": 5, "method": "tools/call", "params": {"_meta": meta, "name": "delete_account"}})
    print(json.dumps(unknown))
    spec = mcp_to_chat_spec(listed["result"]["tools"][0], "brightlane")
    print("as a chat tool spec:", spec["function"]["name"])

  {"jsonrpc": "2.0", "id": 1, "result": {"resultType": "complete", "supportedVersions": ["2026-07-28"], "capabilities": {"tools": {"listChanged": false}}, "instructions": "Brightlane support tools. issue_refund needs human approval in the host."}}
    search_kb          readOnly=True  destructive=False
    read_article       readOnly=True  destructive=False
    get_account        readOnly=True  destructive=False
    get_invoice        readOnly=True  destructive=False
    issue_refund       readOnly=False destructive=True
    add_internal_note  readOnly=False destructive=False
    escalate_to_human  readOnly=False destructive=False
  {"jsonrpc": "2.0", "id": 3, "result": {"resultType": "complete", "content": [{"type": "text", "text": "[{\"article_id\": \"billing-refunds\", \"title\": \"Refunds and cancellations\", \"score\": 4.044}]"}], "structuredCo
  {"jsonrpc": "2.0", "id": 4, "result": {"resultType": "complete", "content": [{"type": "text", "text": "ValueError: No invoice 'INV-2026-999999'"}], "isError": true}}
  {"jsonrpc": "2.0", "id": 5, "error": {"code": -32602, "message": "Unknown tool: delete_account"}}
  as a chat tool spec: brightlane__search_kb

"""Module 8: code execution and workspace tools, with limits, and a demonstration of what the limits do NOT stop.

Linux or macOS only (uses the `resource` module). This is a teaching sandbox: it
bounds time, memory, output, and file size. It is NOT an isolation boundary.
"""
from __future__ import annotations

import os
import resource
import subprocess
import sys
import tempfile
import time
from pathlib import Path


def run_python(code: str, timeout_s: float = 2.0, cpu_s: int = 2, mem_mb: int = 256,
               file_mb: int = 1, max_output: int = 2000) -> dict:
    """Run untrusted Python in a child process with resource limits. Returns a result dict; never raises."""
    def limit_child() -> None:  # runs in the child between fork and exec
        resource.setrlimit(resource.RLIMIT_CPU, (cpu_s, cpu_s))
        resource.setrlimit(resource.RLIMIT_AS, (mem_mb * 2**20, mem_mb * 2**20))
        resource.setrlimit(resource.RLIMIT_FSIZE, (file_mb * 2**20, file_mb * 2**20))
        os.setsid()  # own process group, so a timeout kill takes children with it

    with tempfile.TemporaryDirectory(prefix="agent-sbx-") as tmp:
        started = time.perf_counter()
        try:
            proc = subprocess.run([sys.executable, "-I", "-S", "-c", code], cwd=tmp, env={"PATH": "/usr/bin:/bin"},
                                  capture_output=True, text=True, timeout=timeout_s, preexec_fn=limit_child)
            outcome = "ok" if proc.returncode == 0 else f"exit {proc.returncode}"
            if proc.returncode < 0:
                outcome = f"killed by signal {-proc.returncode}"
            stdout, stderr = proc.stdout, proc.stderr
        except subprocess.TimeoutExpired as exc:
            outcome = f"timeout after {timeout_s} s"
            stdout = (exc.stdout or b"").decode() if isinstance(exc.stdout, bytes) else (exc.stdout or "")
            stderr = ""
        ms = round((time.perf_counter() - started) * 1000)
    return {"outcome": outcome, "ms": ms, "stdout": stdout[:max_output],
            "stderr_tail": stderr.strip().splitlines()[-1][:160] if stderr.strip() else ""}


class Workspace:
    """File tools confined to one directory: the agent's scratch area for drafts and exports."""

    def __init__(self, root: str | Path, max_bytes: int = 100_000) -> None:
        self.root = Path(root).resolve()
        self.root.mkdir(parents=True, exist_ok=True)
        self.max_bytes = max_bytes

    def _path(self, relative: str) -> Path:
        p = (self.root / relative).resolve()  # resolve() follows symlinks and removes ..
        if not p.is_relative_to(self.root):
            raise PermissionError(f"{relative!r} is outside the workspace")
        return p

    def write_file(self, path: str, content: str) -> dict:
        if len(content.encode()) > self.max_bytes:
            raise ValueError(f"content over {self.max_bytes} bytes")
        p = self._path(path)
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(content, encoding="utf-8")
        return {"written": str(p.relative_to(self.root)), "bytes": len(content.encode())}

    def read_file(self, path: str) -> dict:
        p = self._path(path)
        return {"path": path, "content": p.read_text(encoding="utf-8")[: self.max_bytes]}

    def list_files(self) -> list[str]:
        return sorted(str(p.relative_to(self.root)) for p in self.root.rglob("*") if p.is_file())


if __name__ == "__main__":
    repo_file = str(Path("data/tickets.jsonl").resolve())
    cases = {
        "proration math": "seats, monthly, annual = 24, 12, 10\n"
                          "print('monthly per year:', seats * monthly * 12)\n"
                          "print('annual per year :', seats * annual * 12)\n"
                          "print('saving        :', seats * (monthly - annual) * 12)",
        "infinite loop": "while True:\n    pass",
        "memory bomb": "x = bytearray(1024 * 1024 * 1024)\nprint(len(x))",
        "disk filler": "open('big.bin', 'wb').write(b'0' * 5 * 1024 * 1024)",
        "reads outside cwd": f"print(open({repo_file!r}).readline()[:60])",
        "env and network": "import os, socket\nprint(sorted(os.environ))\n"
                           "s = socket.create_connection(('pypi.org', 443), timeout=3)\nprint('connected:', s.getpeername()[1])",
    }
    for name, code in cases.items():
        r = run_python(code)
        print(f"{name:18} {r['outcome']:22} {r['ms']:>5} ms | out: {r['stdout'].strip().replace(chr(10), '; ')[:84]!r} | err: {r['stderr_tail'][:70]!r}")

    ws = Workspace("runs/m08/workspace")
    print("\n", ws.write_file("drafts/T-1001.md", "Hi, we refunded the duplicate charge."))
    print(ws.list_files())
    for bad in ["../../supportdesk/llm.py", "/etc/passwd", "drafts/../../escape.txt"]:
        try:
            ws.read_file(bad)
        except PermissionError as exc:
            print("blocked:", exc)

  proration math     ok                        68 ms | out: 'monthly per year: 3456; annual per year : 2880; saving        : 576' | err: ''
  infinite loop      timeout after 2.0 s     2028 ms | out: '' | err: ''
  memory bomb        exit 1                    49 ms | out: '' | err: 'MemoryError'
  disk filler        exit 1                    67 ms | out: '' | err: 'OSError: [Errno 27] File too large'
  reads outside cwd  ok                        54 ms | out: '{"id": "T-1001", "subject": "Charged twice this month", "bod' | err: ''
  env and network    ok                       107 ms | out: "['LC_CTYPE', 'PATH']; connected: 443" | err: ''

   {'written': 'drafts/T-1001.md', 'bytes': 37}
  ['drafts/T-1001.md']
  blocked: '../../supportdesk/llm.py' is outside the workspace
  blocked: '/etc/passwd' is outside the workspace
  blocked: 'drafts/../../escape.txt' is outside the workspace