MCP API¶

from capiscio_sdk._rpc.client import CapiscioRPCClient

# Connect to capiscio-core
client = CapiscioRPCClient()
client.connect()

# Evaluate tool access for an incoming request
result = client.mcp.evaluate_tool_access(
    tool_name="write_file",
    badge_jws=incoming_badge,
    server_origin="https://files.example.com",
    min_trust_level=1,  # Require at least DV
)

if result["decision"] == "allow":
    print(f"✅ Access granted for agent {result['agent_did']}")
else:
    print(f"❌ Access denied: {result['deny_reason']}")

from capiscio_sdk._rpc.client import CapiscioRPCClient

client = CapiscioRPCClient()
client.connect()

# Access MCP client
mcp = client.mcp

# Now use MCP methods
result = mcp.evaluate_tool_access(tool_name="read_file")

def evaluate_tool_access(
    tool_name: str,
    params_hash: str = "",
    server_origin: str = "",
    *,
    badge_jws: Optional[str] = None,
    api_key: Optional[str] = None,
    policy_version: str = "",
    trusted_issuers: Optional[list[str]] = None,
    min_trust_level: int = 0,
    accept_level_zero: bool = False,
    allowed_tools: Optional[list[str]] = None,
) -> dict

import hashlib
import json

# Evaluate with badge authentication
result = client.mcp.evaluate_tool_access(
    tool_name="write_file",
    params_hash=hashlib.sha256(json.dumps({"path": "/tmp/test.txt"}).encode()).hexdigest(),
    server_origin="https://files.example.com",
    badge_jws=badge_token,
    min_trust_level=2,  # Require OV or higher
    trusted_issuers=["https://registry.capisc.io"],
)

if result["decision"] == "allow":
    # Proceed with tool execution
    execute_tool(tool_name, params)
else:
    raise PermissionError(f"{result['deny_reason']}: {result['deny_detail']}")

def verify_server_identity(
    server_did: str,
    server_badge: str = "",
    transport_origin: str = "",
    endpoint_path: str = "",
    *,
    trusted_issuers: Optional[list[str]] = None,
    min_trust_level: int = 0,
    accept_level_zero: bool = False,
    offline_mode: bool = False,
    skip_origin_binding: bool = False,
) -> dict

# Verify server before trusting tool results
result = client.mcp.verify_server_identity(
    server_did="did:web:files.example.com:mcp:files",
    server_badge=server_badge_token,
    transport_origin="https://files.example.com",
    min_trust_level=1,  # Require at least DV
)

if result["state"] == "verified_principal":
    print(f"✅ Server verified at trust level {result['trust_level']}")
    # Trust the server's responses
else:
    print(f"⚠️ Server not fully verified: {result['error_detail']}")
    # Prompt user for confirmation or reject

def parse_server_identity_http(
    capiscio_server_did: str = "",
    capiscio_server_badge: str = "",
) -> dict

import requests

# Make request to MCP server
response = requests.post("https://files.example.com/mcp/tools/read_file", ...)

# Parse identity from response headers
identity = client.mcp.parse_server_identity_http(
    capiscio_server_did=response.headers.get("Capiscio-Server-DID", ""),
    capiscio_server_badge=response.headers.get("Capiscio-Server-Badge", ""),
)

if identity["identity_present"]:
    # Verify the server identity
    verification = client.mcp.verify_server_identity(
        server_did=identity["server_did"],
        server_badge=identity["server_badge"],
        transport_origin="https://files.example.com",
    )

def parse_server_identity_jsonrpc(meta_json: str) -> dict

import json

# JSON-RPC response from stdio MCP server
response = {
    "jsonrpc": "2.0",
    "id": 1,
    "result": {"content": "..."},
    "_meta": {
        "serverDid": "did:web:files.example.com:mcp:files",
        "serverBadge": "eyJhbGciOiJFZERTQSI..."
    }
}

# Parse identity from _meta
identity = client.mcp.parse_server_identity_jsonrpc(
    meta_json=json.dumps(response.get("_meta", {}))
)

if identity["identity_present"]:
    # Verify the server
    verification = client.mcp.verify_server_identity(
        server_did=identity["server_did"],
        server_badge=identity["server_badge"],
    )

def health(client_version: str = "") -> dict

status = client.mcp.health(client_version="capiscio-sdk-python/2.5.0")

if status["healthy"]:
    print(f"✅ MCP service healthy (core {status['core_version']})")
else:
    print("❌ MCP service unhealthy")

# Require different trust levels for different tools
sensitive_tools = ["execute_shell", "delete_file", "modify_system"]

if tool_name in sensitive_tools:
    min_level = 3  # Require OV for dangerous tools
else:
    min_level = 2  # DV sufficient for normal tools (Level 2)

result = client.mcp.evaluate_tool_access(
    tool_name=tool_name,
    badge_jws=badge,
    min_trust_level=min_level,
)

from fastapi import FastAPI, Request, HTTPException
from capiscio_sdk._rpc.client import CapiscioRPCClient

app = FastAPI()
client = CapiscioRPCClient()
client.connect()

@app.middleware("http")
async def mcp_auth(request: Request, call_next):
    # Skip auth for health/docs
    if request.url.path in ["/health", "/docs", "/openapi.json"]:
        return await call_next(request)

    # Extract badge from Authorization header
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        raise HTTPException(401, "Missing badge token")

    badge = auth[7:]

    # Extract tool name from path (e.g., /tools/read_file)
    tool_name = request.url.path.split("/")[-1]

    # Evaluate access
    result = client.mcp.evaluate_tool_access(
        tool_name=tool_name,
        badge_jws=badge,
        server_origin=str(request.base_url).rstrip("/"),
        min_trust_level=1,
        trusted_issuers=["https://registry.capisc.io"],
    )

    if result["decision"] != "allow":
        raise HTTPException(
            403,
            detail={
                "error": "access_denied",
                "reason": result["deny_reason"],
                "detail": result["deny_detail"],
            }
        )

    # Attach agent info to request
    request.state.agent_did = result["agent_did"]
    request.state.trust_level = result["trust_level"]

    return await call_next(request)

from capiscio_sdk._rpc.client import CapiscioRPCClient
import requests

client = CapiscioRPCClient()
client.connect()

def call_mcp_tool(server_url: str, tool_name: str, params: dict) -> dict:
    """Call an MCP tool with server identity verification."""

    # Make the tool call
    response = requests.post(
        f"{server_url}/tools/{tool_name}",
        json=params,
        headers={"Authorization": f"Bearer {my_badge}"},
    )

    # Parse and verify server identity
    identity = client.mcp.parse_server_identity_http(
        capiscio_server_did=response.headers.get("Capiscio-Server-DID", ""),
        capiscio_server_badge=response.headers.get("Capiscio-Server-Badge", ""),
    )

    if not identity["identity_present"]:
        raise ValueError("Server did not provide identity headers")

    verification = client.mcp.verify_server_identity(
        server_did=identity["server_did"],
        server_badge=identity["server_badge"],
        transport_origin=server_url,
        min_trust_level=1,
    )

    if verification["state"] != "verified_principal":
        raise ValueError(f"Server verification failed: {verification['error_detail']}")

    return response.json()

# Define allowed tools per trust level
TOOL_ALLOWLIST = {
    0: ["read_file", "list_directory"],  # Level 0: read-only
    1: ["read_file", "list_directory", "search"],  # Level 1: + search
    2: ["read_file", "list_directory", "search", "write_file"],  # Level 2: + write
    3: ["read_file", "list_directory", "search", "write_file", "execute"],  # Level 3: + execute
}

def get_allowed_tools(trust_level: int) -> list[str]:
    """Get allowed tools for a trust level."""
    return TOOL_ALLOWLIST.get(trust_level, TOOL_ALLOWLIST[0])

# Use in evaluation
result = client.mcp.evaluate_tool_access(
    tool_name=requested_tool,
    badge_jws=badge,
    allowed_tools=get_allowed_tools(caller_trust_level),
)

from grpc import RpcError, StatusCode

try:
    result = client.mcp.evaluate_tool_access(
        tool_name="write_file",
        badge_jws=badge,
    )
except RpcError as e:
    if e.code() == StatusCode.UNAVAILABLE:
        print("MCP service unavailable - check capiscio-core is running")
    elif e.code() == StatusCode.INVALID_ARGUMENT:
        print(f"Invalid request: {e.details()}")
    else:
        raise