Prism MCP

use Prism\Mcp\Facades\PrismMcp;
use Prism\Prism\Facades\Prism;

$tools = PrismMcp::server('github')->tools();

Prism::text()
    ->using('anthropic', 'claude-sonnet-4-5')
    ->withTools($tools)
    ->withPrompt('Which of our open PRs touch the billing module?')
    ->asText();

import { Client, TrustPolicy } from '@particle-academy/prism-mcp';
import { httpTransport } from './mcp-http'; // see "HTTP transport" below

const github = new Client({
  server: 'github',
  transport: httpTransport(process.env.MCP_GITHUB_URL!),
  trust: TrustPolicy.allowing(['search_repositories']),
});

const [search] = await github.listTools();
const result = await github.callTool(search, { query: 'billing module' });

import os

from prism_mcp import Client, TrustPolicy

from mcp_http import http_transport  # see "HTTP transport" below

github = Client(
    "github",
    http_transport(os.environ["MCP_GITHUB_URL"]),
    trust=TrustPolicy.allowing(["search_repositories"]),
)

search = (await github.list_tools())[0]
result = await github.call_tool(search, {"query": "billing module"})

composer require particle-academy/prism-mcp
php artisan vendor:publish --tag=prism-mcp-config

npm install @particle-academy/prism-mcp

pip install prism-ai-mcp

// config/prism-mcp.php
'servers' => [
    'github' => [
        'transport' => 'http',
        'url' => env('MCP_GITHUB_URL'),
        'headers' => [
            'Authorization' => 'Bearer '.env('MCP_GITHUB_TOKEN', ''),
        ],
        'trust' => [
            'tools' => ['search_repositories', 'get_file_contents'],
        ],
    ],
],

const github = new Client({
  server: 'github',
  transport,
  trust: TrustPolicy.allowing(['search_repositories', 'get_file_contents']),
});

github = Client(
    "github",
    transport,
    trust=TrustPolicy.allowing(["search_repositories", "get_file_contents"]),
)

'trust' => [
    'tools' => ['search_repositories'],
    'pins' => ['search_repositories' => 'sha256:…'],
],

TrustPolicy.allowing(
  ['search_repositories'],
  { search_repositories: 'sha256:…' },
);

TrustPolicy.allowing(
    ["search_repositories"],
    pins={"search_repositories": "sha256:…"},
)

const inspect = new Client({ server: 'github', transport, trust: TrustPolicy.allowingEveryTool() });

for (const tool of await inspect.listTools()) {
  console.log(tool.name, tool.digest(), tool.description);
}

inspect = Client("github", transport, trust=TrustPolicy.allowing_every_tool())

for tool in await inspect.list_tools():
    print(tool.name, tool.digest(), tool.description)

$tools = PrismMcp::client('https://mcp.example.com/mcp', ['name' => 'example'])
    ->withToken(config('services.example.token'))
    ->trusting(['search'])
    ->withTimeout(10)
    ->tools();

const client = new Client({
  server: 'example',
  transport: httpTransport('https://mcp.example.com/mcp', {
    headers: { Authorization: `Bearer ${process.env.EXAMPLE_TOKEN}` },
    timeoutMs: 10_000,
  }),
  trust: TrustPolicy.allowing(['search']),
});

const tools = await client.listTools();

client = Client(
    "example",
    http_transport(
        "https://mcp.example.com/mcp",
        headers={"Authorization": f"Bearer {os.environ['EXAMPLE_TOKEN']}"},
        timeout=10.0,
    ),
    trust=TrustPolicy.allowing(["search"]),
)

tools = await client.list_tools()

$tools = PrismMcp::server('github')
    ->withMaxResultBytes(128 * 1024)
    ->filteringResults(fn (string $text, string $server, string $tool): string => trim($text))
    ->tools();

import { ResultGuard } from '@particle-academy/prism-mcp';

const github = new Client({
  server: 'github',
  transport,
  trust: TrustPolicy.allowing(['search_repositories']),
  guard: new ResultGuard({
    maxBytes: 128 * 1024,
    frameProvenance: true,
    filter: (server, tool, text) => text.trim(),
  }),
});

from prism_mcp import ResultGuard

github = Client(
    "github",
    transport,
    trust=TrustPolicy.allowing(["search_repositories"]),
    guard=ResultGuard(
        max_bytes=128 * 1024,
        frame_provenance=True,
        filter=lambda server, tool, text: text.strip(),
    ),
)

import { type ToolGate } from '@particle-academy/prism-mcp';

const gate: ToolGate = (server, tool) => currentUser.can(`mcp:${server}:${tool}`);

const github = new Client({ server: 'github', transport, trust, gate });

def gate(server: str, tool: str, args: dict[str, object]) -> bool:
    return current_user.can(f"mcp:{server}:{tool}")

github = Client("github", transport, trust=trust, gate=gate)

// mcp-http.ts
import {
  LATEST_PROTOCOL_VERSION,
  McpError,
  isJsonObject,
  type JsonObject,
  type JsonValue,
  type Transport,
} from '@particle-academy/prism-mcp';

export interface HttpTransportOptions {
  headers?: Record<string, string>;
  timeoutMs?: number;
  maxResponseBytes?: number;
  clientInfo?: { name: string; version: string };
}

// Header values can carry model-supplied text. Anything that is not plain,
// trimmed, printable ASCII travels base64 in the sentinel the spec defines.
function headerValue(value: string): string {
  const plain = /^[\x20-\x7E]*$/.test(value) && value === value.trim() && !value.startsWith('=?');

  return plain ? value : `=?base64?${Buffer.from(value, 'utf8').toString('base64')}?=`;
}

async function readCapped(response: Response, limit: number): Promise<string> {
  const reader = response.body?.getReader();
  if (reader === undefined) return '';

  const chunks: Uint8Array[] = [];
  let size = 0;

  for (;;) {
    const { done, value } = await reader.read();
    if (done) break;
    size += value.byteLength;
    if (size > limit) {
      await reader.cancel();
      throw new McpError('protocol_failure', `The MCP server sent more than ${limit} bytes.`);
    }
    chunks.push(value);
  }

  return Buffer.concat(chunks).toString('utf8');
}

// A server may answer with an event stream; the LAST JSON-RPC frame is the result.
function lastFrame(body: string): string {
  const frames = body
    .split(/\r\n|\r|\n/)
    .filter((line) => line.startsWith('data:'))
    .map((line) => line.slice(5).trim())
    .filter((data) => data.includes('"jsonrpc"'));

  const last = frames.at(-1);
  if (last === undefined) throw new McpError('protocol_failure', 'The event stream carried no JSON-RPC frame.');

  return last;
}

export function httpTransport(url: string, options: HttpTransportOptions = {}): Transport {
  const clientInfo = options.clientInfo ?? { name: 'my-app', version: '1.0.0' };
  let nextId = 1;

  return async ({ method, params = {}, headers = {} }) => {
    const id = nextId++;
    const meta = isJsonObject(params._meta) ? params._meta : {};

    // 2026-07-28 is stateless: every request carries its own version, capabilities and client info.
    const body = {
      jsonrpc: '2.0',
      id,
      method,
      params: {
        ...params,
        _meta: {
          'io.modelcontextprotocol/protocolVersion': LATEST_PROTOCOL_VERSION,
          'io.modelcontextprotocol/clientCapabilities': {},
          'io.modelcontextprotocol/clientInfo': clientInfo,
          ...meta,
        },
      },
    };

    const requestHeaders: Record<string, string> = {
      ...options.headers,
      'Content-Type': 'application/json',
      Accept: 'application/json, text/event-stream',
      'MCP-Protocol-Version': LATEST_PROTOCOL_VERSION,
      'Mcp-Method': method,
    };

    if (typeof params.name === 'string') requestHeaders['Mcp-Name'] = headerValue(params.name);

    // The client passes the tool's mirrored parameters (`Mcp-Param-*`) here.
    for (const [name, value] of Object.entries(headers)) requestHeaders[name] = headerValue(value);

    const response = await fetch(url, {
      method: 'POST',
      headers: requestHeaders,
      body: JSON.stringify(body),
      signal: AbortSignal.timeout(options.timeoutMs ?? 30_000),
    });

    if (!response.ok) {
      throw new McpError('protocol_failure', `The MCP server answered HTTP ${response.status}.`);
    }

    const text = await readCapped(response, options.maxResponseBytes ?? 8 * 1024 * 1024);
    const raw = (response.headers.get('content-type') ?? '').includes('text/event-stream') ? lastFrame(text) : text;

    let envelope: JsonValue;
    try {
      envelope = JSON.parse(raw) as JsonValue;
    } catch {
      throw new McpError('protocol_failure', 'The MCP server did not answer with JSON.');
    }

    if (!isJsonObject(envelope) || envelope.jsonrpc !== '2.0' || envelope.id !== id) {
      throw new McpError('protocol_failure', `The response is not the JSON-RPC answer to request ${id}.`);
    }

    if (isJsonObject(envelope.error)) {
      const error: JsonObject = envelope.error;
      throw new McpError(
        error.code === -32022 ? 'unsupported_protocol_version' : 'protocol_failure',
        `JSON-RPC error ${String(error.code)}: ${String(error.message)}`,
      );
    }

    return envelope.result ?? null;
  };
}

# mcp_http.py
from __future__ import annotations

import asyncio
import base64
import itertools
import json
import re
import urllib.error
import urllib.request
from typing import Any

from prism_mcp import LATEST_PROTOCOL_VERSION, ErrorCode, McpError, Transport, TransportRequest

_PLAIN = re.compile(r"[\x20-\x7e]*")


def _header_value(value: str) -> str:
    # Header values can carry model-supplied text. Anything that is not plain,
    # trimmed, printable ASCII travels base64 in the sentinel the spec defines.
    if _PLAIN.fullmatch(value) and value == value.strip() and not value.startswith("=?"):
        return value
    return "=?base64?" + base64.b64encode(value.encode("utf-8")).decode("ascii") + "?="


def _last_frame(body: str) -> str:
    # A server may answer with an event stream; the LAST JSON-RPC frame is the result.
    frames = [
        line[5:].strip()
        for line in body.splitlines()
        if line.startswith("data:") and '"jsonrpc"' in line
    ]
    if not frames:
        raise McpError(ErrorCode.PROTOCOL_FAILURE, "The event stream carried no JSON-RPC frame.")
    return frames[-1]


def http_transport(
    url: str,
    headers: dict[str, str] | None = None,
    timeout: float = 30.0,
    max_response_bytes: int = 8 * 1024 * 1024,
    client_info: dict[str, str] | None = None,
) -> Transport:
    info = client_info or {"name": "my-app", "version": "1.0.0"}
    ids = itertools.count(1)

    def send(request: TransportRequest, request_id: int) -> Any:
        params = dict(request.params or {})
        existing = params.get("_meta")
        meta: dict[str, Any] = existing if isinstance(existing, dict) else {}
        # 2026-07-28 is stateless: every request carries its own version, capabilities and client info.
        params["_meta"] = {
            "io.modelcontextprotocol/protocolVersion": LATEST_PROTOCOL_VERSION,
            "io.modelcontextprotocol/clientCapabilities": {},
            "io.modelcontextprotocol/clientInfo": info,
            **meta,
        }
        body = {"jsonrpc": "2.0", "id": request_id, "method": request.method, "params": params}

        sent = {
            **(headers or {}),
            "Content-Type": "application/json",
            "Accept": "application/json, text/event-stream",
            "MCP-Protocol-Version": LATEST_PROTOCOL_VERSION,
            "Mcp-Method": request.method,
        }
        if isinstance(params.get("name"), str):
            sent["Mcp-Name"] = _header_value(params["name"])
        # The client passes the tool's mirrored parameters (Mcp-Param-*) here.
        for name, value in request.headers.items():
            sent[name] = _header_value(value)

        http_request = urllib.request.Request(
            url, data=json.dumps(body).encode("utf-8"), headers=sent, method="POST"
        )
        try:
            with urllib.request.urlopen(http_request, timeout=timeout) as response:
                raw_bytes = response.read(max_response_bytes + 1)
                content_type = response.headers.get("Content-Type", "")
        except urllib.error.HTTPError as failure:
            raise McpError(
                ErrorCode.PROTOCOL_FAILURE, f"The MCP server answered HTTP {failure.code}."
            ) from failure

        if len(raw_bytes) > max_response_bytes:
            raise McpError(
                ErrorCode.PROTOCOL_FAILURE,
                f"The MCP server sent more than {max_response_bytes} bytes.",
            )

        text = raw_bytes.decode("utf-8")
        raw = _last_frame(text) if "text/event-stream" in content_type else text

        try:
            envelope = json.loads(raw)
        except ValueError as failure:
            raise McpError(
                ErrorCode.PROTOCOL_FAILURE, "The MCP server did not answer with JSON."
            ) from failure

        if (
            not isinstance(envelope, dict)
            or envelope.get("jsonrpc") != "2.0"
            or envelope.get("id") != request_id
        ):
            raise McpError(
                ErrorCode.PROTOCOL_FAILURE,
                f"The response is not the JSON-RPC answer to request {request_id}.",
            )

        error = envelope.get("error")
        if isinstance(error, dict):
            code = (
                ErrorCode.UNSUPPORTED_PROTOCOL_VERSION
                if error.get("code") == -32022
                else ErrorCode.PROTOCOL_FAILURE
            )
            raise McpError(code, f"JSON-RPC error {error.get('code')}: {error.get('message')}")

        return envelope.get("result")

    async def transport(request: TransportRequest) -> Any:
        # urllib blocks, so the request runs in a worker thread.
        return await asyncio.to_thread(send, request, next(ids))

    return transport