API Reference¶
This section provides detailed API documentation for all public modules in capiscio-mcp.
Core Exports¶
capiscio-mcp: Trust badges for MCP tool calls.
RFC-006: MCP Tool Authority and Evidence RFC-007: MCP Server Identity Disclosure and Verification
This package provides: - @guard decorator for protecting MCP tools with trust-level requirements - Server identity verification for MCP clients - Server identity registration for MCP servers - PoP (Proof of Possession) handshake for server key verification - Evidence logging for audit and forensics
Installation
pip install capiscio-mcp # Standalone pip install capiscio-mcp[mcp] # With MCP SDK integration pip install capiscio-mcp[crypto] # With PoP signing/verification
Quickstart (Server-side): from capiscio_mcp import guard
@guard(min_trust_level=2)
async def read_database(query: str) -> list[dict]:
...
Quickstart (Client-side): from capiscio_mcp import verify_server, ServerState
result = await verify_server(
server_did="did:web:mcp.example.com",
server_badge="eyJhbGc...",
)
if result.state == ServerState.VERIFIED_PRINCIPAL:
print(f"Trusted at level {result.trust_level}")
Quickstart (Server Registration): from capiscio_mcp import setup_server_identity
result = await setup_server_identity(
server_id="your-server-uuid",
api_key="sk_live_...",
output_dir="./keys",
)
print(f"Server DID: {result['did']}")
GuardConfig dataclass ¶
Configuration for the @guard decorator.
Attributes:
| Name | Type | Description |
|---|---|---|
min_trust_level | int | Minimum trust level required (0-4, default 0) |
accept_level_zero | bool | Accept self-signed (did:key) badges |
trusted_issuers | Optional[List[str]] | List of trusted issuer DIDs |
allowed_tools | Optional[List[str]] | Glob patterns for allowed tool names |
policy_version | Optional[str] | Policy version string for tracking |
require_badge | bool | If True, deny anonymous/API key access |
GuardResult dataclass ¶
Result from tool access evaluation.
Attributes:
| Name | Type | Description |
|---|---|---|
decision | Decision | ALLOW or DENY |
deny_reason | Optional[DenyReason] | Reason for denial (if decision is DENY) |
deny_detail | Optional[str] | Human-readable detail (if decision is DENY) |
agent_did | Optional[str] | Extracted agent DID from credential |
badge_jti | Optional[str] | Badge ID if present |
auth_level | AuthLevel | Authentication level (ANONYMOUS, API_KEY, BADGE) |
trust_level | int | Verified trust level (0-4) |
evidence_id | str | Unique evidence record ID |
evidence_json | str | RFC-006 §7 compliant JSON |
GuardError ¶
Bases: CapiscioMCPError
Raised when tool access is denied by the guard.
Per RFC-006, all denials include: - reason: Specific denial reason code - detail: Human-readable explanation - evidence_id: ID of the evidence record for audit
Example
try: result = await guarded_tool(params) except GuardError as e: logger.warning(f"Access denied: {e.reason} - {e.detail}") logger.info(f"Evidence ID: {e.evidence_id}")
VerifyConfig dataclass ¶
Configuration for server identity verification.
Attributes:
| Name | Type | Description |
|---|---|---|
trusted_issuers | Optional[List[str]] | List of trusted issuer DIDs |
min_trust_level | int | Minimum trust level required (0-4) |
accept_level_zero | bool | Accept self-signed (did:key) servers |
offline_mode | bool | Skip revocation checks |
skip_origin_binding | bool | Skip host/path binding checks (for trusted gateways) |
VerifyResult dataclass ¶
Result of server identity verification.
Attributes:
| Name | Type | Description |
|---|---|---|
state | ServerState | Server classification state (VERIFIED_PRINCIPAL, DECLARED_PRINCIPAL, UNVERIFIED_ORIGIN) |
trust_level | Optional[int] | Trust level if verified (0-4) |
server_did | Optional[str] | Server DID if disclosed |
badge_jti | Optional[str] | Badge ID if present |
error_code | ServerErrorCode | Error code if verification failed |
error_detail | Optional[str] | Human-readable error detail |
Decision ¶
Bases: str, Enum
Tool access decision result.
Per RFC-006 §6.3, every tool invocation attempt results in one of: - ALLOW: Tool execution is permitted - DENY: Tool execution is blocked with a reason
AuthLevel ¶
Bases: str, Enum
Caller authentication assurance level.
Per RFC-006 §5, every evidence log records the authentication method: - ANONYMOUS: No identity material provided - API_KEY: API key authentication (reduced assurance) - BADGE: CapiscIO Trust Badge authentication (full assurance)
DenyReason ¶
Bases: str, Enum
Reason for denying tool access.
Per RFC-006 §6.4, denial must include a specific reason code.
ServerState ¶
Bases: str, Enum
Server identity verification state.
Per RFC-007 §5.2, clients classify servers into three states:
-
VERIFIED_PRINCIPAL: Server badge verified, trust level established. The server has disclosed a DID and a valid badge signed by a trusted issuer.
-
DECLARED_PRINCIPAL: Server DID disclosed but badge missing or invalid. Identity is claimed but not cryptographically verified.
-
UNVERIFIED_ORIGIN: Server did not disclose any identity material. This is distinct from Trust Level 0 (self-signed) - UNVERIFIED_ORIGIN means NO identity was disclosed at all.
ServerErrorCode ¶
Bases: str, Enum
Server identity verification error codes.
Per RFC-007 §8, verification failures include specific error codes.
RegistrationError ¶
Bases: Exception
Error during server identity registration.
KeyGenerationError ¶
Bases: Exception
Error generating keypair.
guard_sync ¶
guard_sync(
func: Optional[Callable[P, R]] = None,
*,
config: Optional[GuardConfig] = None,
min_trust_level: Optional[int] = None,
tool_name: Optional[str] = None,
require_badge: bool = False
) -> Union[
Callable[P, R],
Callable[[Callable[P, R]], Callable[P, R]],
]
Sync decorator to guard MCP tool execution.
Same as @guard but for synchronous functions. Internally runs the async guard in an event loop.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
func | Optional[Callable[P, R]] | Function to decorate (if called without parentheses) | None |
config | Optional[GuardConfig] | Full configuration object | None |
min_trust_level | Optional[int] | Shorthand for config.min_trust_level | None |
tool_name | Optional[str] | Override tool name (default: function name) | None |
require_badge | bool | If True, deny anonymous/API key access | False |
Example
@guard_sync(min_trust_level=2) def read_file(path: str) -> str: ...
verify_server async ¶
verify_server(
server_did: Optional[str],
server_badge: Optional[str] = None,
transport_origin: Optional[str] = None,
endpoint_path: Optional[str] = None,
config: Optional[VerifyConfig] = None,
) -> VerifyResult
Verify MCP server identity per RFC-007 §7.2.
This function implements the client verification algorithm: 1. If no DID disclosed → UNVERIFIED_ORIGIN 2. If DID but no badge → DECLARED_PRINCIPAL 3. If DID + badge → verify badge → VERIFIED_PRINCIPAL or error
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
server_did | Optional[str] | Server DID from Capiscio-Server-DID header or _meta | required |
server_badge | Optional[str] | Server badge JWS from Capiscio-Server-Badge header or _meta | None |
transport_origin | Optional[str] | HTTP origin (e.g., "https://mcp.example.com") | None |
endpoint_path | Optional[str] | URL path for did:web binding (e.g., "/mcp/filesystem") | None |
config | Optional[VerifyConfig] | Verification configuration | None |
Returns:
| Type | Description |
|---|---|
VerifyResult | VerifyResult with state, trust_level, and any error details |
Example
result = await verify_server( server_did="did:web:mcp.example.com:servers:filesystem", server_badge="eyJhbGciOiJFZERTQSIsInR5cCI6IkpXVCJ9...", transport_origin="https://mcp.example.com", )
match result.state: case ServerState.VERIFIED_PRINCIPAL: print(f"Verified at level {result.trust_level}") case ServerState.DECLARED_PRINCIPAL: print("Identity claimed but not verified") case ServerState.UNVERIFIED_ORIGIN: print("No identity disclosed")
verify_server_sync ¶
verify_server_sync(
server_did: Optional[str],
server_badge: Optional[str] = None,
transport_origin: Optional[str] = None,
endpoint_path: Optional[str] = None,
config: Optional[VerifyConfig] = None,
) -> VerifyResult
Sync wrapper for verify_server.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
server_did | Optional[str] | Server DID from Capiscio-Server-DID header or _meta | required |
server_badge | Optional[str] | Server badge JWS from Capiscio-Server-Badge header or _meta | None |
transport_origin | Optional[str] | HTTP origin (e.g., "https://mcp.example.com") | None |
endpoint_path | Optional[str] | URL path for did:web binding | None |
config | Optional[VerifyConfig] | Verification configuration | None |
Returns:
| Type | Description |
|---|---|
VerifyResult | VerifyResult with state, trust_level, and any error details |
generate_server_keypair async ¶
Generate Ed25519 keypair for MCP server identity.
Uses capiscio-core's SimpleGuardService.GenerateKeyPair via gRPC. The keypair is used for PoP (Proof of Possession) verification.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key_id | str | Optional specific key ID. If empty, one is generated. | '' |
output_dir | Optional[str] | Optional directory to save private key PEM file. If provided, saves as {key_id}.pem | None |
Returns:
| Type | Description |
|---|---|
dict | dict with: - key_id: The key identifier - did_key: The derived did:key URI (e.g., did |
Raises:
| Type | Description |
|---|---|
KeyGenerationError | If key generation fails |
CoreConnectionError | If connection to capiscio-core fails |
Example
keys = await generate_server_keypair(output_dir="./keys") print(f"DID: {keys['did_key']}")
did
z6MkhaXgBZD...¶
generate_server_keypair_sync ¶
Sync wrapper for generate_server_keypair.
See generate_server_keypair() for full documentation.
register_server_identity async ¶
register_server_identity(
server_id: str,
api_key: str,
did: str,
public_key: str,
ca_url: str = "https://registry.capisc.io",
) -> dict
Register MCP server DID with the CapiscIO registry.
Uses PUT /v1/sdk/servers/{id} to update the server's DID and public key. This follows the same pattern as agent identity registration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
server_id | str | The MCP server's UUID (from dashboard creation) | required |
api_key | str | Registry API key (X-Capiscio-Registry-Key) | required |
did | str | The server's DID (e.g., did | required |
public_key | str | PEM-encoded public key | required |
ca_url | str | Registry URL (default: https://registry.capisc.io) | 'https://registry.capisc.io' |
Returns:
| Type | Description |
|---|---|
dict | dict with: - success: True if registration succeeded - message: Status message - data: Updated server object (if successful) |
Raises:
| Type | Description |
|---|---|
RegistrationError | If registration fails |
Example
result = await register_server_identity( server_id="550e8400-e29b-41d4-a716-446655440000", api_key="sk_live_abc123...", did="didz6MkhaXgBZD...", public_key="-----BEGIN PUBLIC KEY-----...", )
register_server_identity_sync ¶
register_server_identity_sync(
server_id: str,
api_key: str,
did: str,
public_key: str,
ca_url: str = "https://registry.capisc.io",
) -> dict
Sync wrapper for register_server_identity.
See register_server_identity() for full documentation.
setup_server_identity async ¶
setup_server_identity(
server_id: str,
api_key: str,
ca_url: str = "https://registry.capisc.io",
output_dir: Optional[str] = None,
key_id: str = "",
) -> dict
Generate keypair and register server identity in one call.
This is the recommended way to set up MCP server identity: 1. Generates Ed25519 keypair via capiscio-core 2. Registers the DID with the CapiscIO registry 3. Optionally saves the private key to disk
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
server_id | str | The MCP server's UUID (from dashboard creation) | required |
api_key | str | Registry API key (X-Capiscio-Registry-Key) | required |
ca_url | str | Registry URL (default: https://registry.capisc.io) | 'https://registry.capisc.io' |
output_dir | Optional[str] | Optional directory to save private key PEM file | None |
key_id | str | Optional specific key ID. If empty, one is generated. | '' |
Returns:
| Type | Description |
|---|---|
dict | dict with: - did: The server's DID (did |
Raises:
| Type | Description |
|---|---|
KeyGenerationError | If key generation fails |
RegistrationError | If registry registration fails |
CoreConnectionError | If connection to capiscio-core fails |
setup_server_identity_sync ¶
setup_server_identity_sync(
server_id: str,
api_key: str,
ca_url: str = "https://registry.capisc.io",
output_dir: Optional[str] = None,
key_id: str = "",
) -> dict
Sync wrapper for setup_server_identity.
See setup_server_identity() for full documentation.
Guard Module (RFC-006)¶
RFC-006: MCP Tool Authority Guard.
This module provides the @guard decorator for protecting MCP tool execution with identity verification and policy enforcement.
Key design principle: params_hash is computed in Python, never sent raw to core. This keeps PII out of the gRPC boundary and avoids cross-language canonicalization.
Usage
from capiscio_mcp import guard, GuardConfig
@guard(min_trust_level=2) async def read_file(path: str) -> str: ...
With full configuration¶
config = GuardConfig( min_trust_level=2, trusted_issuers=["did:web:registry.capisc.io"], allowed_tools=["read_", "list_"], )
@guard(config=config) async def execute_query(sql: str) -> list[dict]: ...
GuardConfig dataclass ¶
Configuration for the @guard decorator.
Attributes:
| Name | Type | Description |
|---|---|---|
min_trust_level | int | Minimum trust level required (0-4, default 0) |
accept_level_zero | bool | Accept self-signed (did:key) badges |
trusted_issuers | Optional[List[str]] | List of trusted issuer DIDs |
allowed_tools | Optional[List[str]] | Glob patterns for allowed tool names |
policy_version | Optional[str] | Policy version string for tracking |
require_badge | bool | If True, deny anonymous/API key access |
GuardResult dataclass ¶
Result from tool access evaluation.
Attributes:
| Name | Type | Description |
|---|---|---|
decision | Decision | ALLOW or DENY |
deny_reason | Optional[DenyReason] | Reason for denial (if decision is DENY) |
deny_detail | Optional[str] | Human-readable detail (if decision is DENY) |
agent_did | Optional[str] | Extracted agent DID from credential |
badge_jti | Optional[str] | Badge ID if present |
auth_level | AuthLevel | Authentication level (ANONYMOUS, API_KEY, BADGE) |
trust_level | int | Verified trust level (0-4) |
evidence_id | str | Unique evidence record ID |
evidence_json | str | RFC-006 §7 compliant JSON |
guard ¶
guard(
func: Optional[
Callable[P, Coroutine[Any, Any, R]]
] = None,
*,
config: Optional[GuardConfig] = None,
min_trust_level: Optional[int] = None,
tool_name: Optional[str] = None,
require_badge: bool = False
) -> Union[
Callable[P, Coroutine[Any, Any, R]],
Callable[
[Callable[P, Coroutine[Any, Any, R]]],
Callable[P, Coroutine[Any, Any, R]],
],
]
Async decorator to guard MCP tool execution.
Single RPC call returns decision + evidence atomically.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
func | Optional[Callable[P, Coroutine[Any, Any, R]]] | Function to decorate (if called without parentheses) | None |
config | Optional[GuardConfig] | Full configuration object | None |
min_trust_level | Optional[int] | Shorthand for config.min_trust_level | None |
tool_name | Optional[str] | Override tool name (default: function name) | None |
require_badge | bool | If True, deny anonymous/API key access | False |
Returns:
| Type | Description |
|---|---|
Union[Callable[P, Coroutine[Any, Any, R]], Callable[[Callable[P, Coroutine[Any, Any, R]]], Callable[P, Coroutine[Any, Any, R]]]] | Decorated function that enforces access control |
Example
Simple usage¶
@guard async def list_files() -> list[str]: ...
With trust level requirement¶
@guard(min_trust_level=2) async def execute_query(sql: str) -> list[dict]: ...
With full configuration¶
@guard(config=GuardConfig( min_trust_level=2, trusted_issuers=["did:web:registry.capisc.io"], )) async def sensitive_operation(data: dict) -> dict: ...
guard_sync ¶
guard_sync(
func: Optional[Callable[P, R]] = None,
*,
config: Optional[GuardConfig] = None,
min_trust_level: Optional[int] = None,
tool_name: Optional[str] = None,
require_badge: bool = False
) -> Union[
Callable[P, R],
Callable[[Callable[P, R]], Callable[P, R]],
]
Sync decorator to guard MCP tool execution.
Same as @guard but for synchronous functions. Internally runs the async guard in an event loop.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
func | Optional[Callable[P, R]] | Function to decorate (if called without parentheses) | None |
config | Optional[GuardConfig] | Full configuration object | None |
min_trust_level | Optional[int] | Shorthand for config.min_trust_level | None |
tool_name | Optional[str] | Override tool name (default: function name) | None |
require_badge | bool | If True, deny anonymous/API key access | False |
Example
@guard_sync(min_trust_level=2) def read_file(path: str) -> str: ...
compute_params_hash ¶
Compute deterministic hash of tool parameters.
CRITICAL: This stays in Python. Core never sees raw params.
Canonicalization rules (JCS-like): 1. Sort keys recursively (lexicographic) 2. Compact JSON (no whitespace) 3. SHA-256 hash 4. Base64url encode (no padding)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
params | dict[str, Any] | Tool parameters dictionary | required |
Returns:
| Type | Description |
|---|---|
str | String in format "sha256: |
Example
compute_params_hash({"b": 2, "a": 1}) 'sha256:...' # Same as compute_params_hash({"a": 1, "b": 2})
Server Module (RFC-007)¶
RFC-007: MCP Server Identity Verification.
This module provides functions to verify MCP server identity before establishing trust with the server.
Key distinction from Trust Level 0: - UNVERIFIED_ORIGIN: Server disclosed NO identity material at all - Trust Level 0: Server disclosed a self-signed (did:key) identity
Usage
from capiscio_mcp import verify_server, ServerState, VerifyConfig
result = await verify_server( server_did="did:web:mcp.example.com:servers:filesystem", server_badge="eyJhbGc...", transport_origin="https://mcp.example.com", )
if result.state == ServerState.VERIFIED_PRINCIPAL: print(f"Server verified at trust level {result.trust_level}") elif result.state == ServerState.UNVERIFIED_ORIGIN: print("Warning: Server did not disclose identity")
VerifyConfig dataclass ¶
Configuration for server identity verification.
Attributes:
| Name | Type | Description |
|---|---|---|
trusted_issuers | Optional[List[str]] | List of trusted issuer DIDs |
min_trust_level | int | Minimum trust level required (0-4) |
accept_level_zero | bool | Accept self-signed (did:key) servers |
offline_mode | bool | Skip revocation checks |
skip_origin_binding | bool | Skip host/path binding checks (for trusted gateways) |
VerifyResult dataclass ¶
Result of server identity verification.
Attributes:
| Name | Type | Description |
|---|---|---|
state | ServerState | Server classification state (VERIFIED_PRINCIPAL, DECLARED_PRINCIPAL, UNVERIFIED_ORIGIN) |
trust_level | Optional[int] | Trust level if verified (0-4) |
server_did | Optional[str] | Server DID if disclosed |
badge_jti | Optional[str] | Badge ID if present |
error_code | ServerErrorCode | Error code if verification failed |
error_detail | Optional[str] | Human-readable error detail |
verify_server async ¶
verify_server(
server_did: Optional[str],
server_badge: Optional[str] = None,
transport_origin: Optional[str] = None,
endpoint_path: Optional[str] = None,
config: Optional[VerifyConfig] = None,
) -> VerifyResult
Verify MCP server identity per RFC-007 §7.2.
This function implements the client verification algorithm: 1. If no DID disclosed → UNVERIFIED_ORIGIN 2. If DID but no badge → DECLARED_PRINCIPAL 3. If DID + badge → verify badge → VERIFIED_PRINCIPAL or error
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
server_did | Optional[str] | Server DID from Capiscio-Server-DID header or _meta | required |
server_badge | Optional[str] | Server badge JWS from Capiscio-Server-Badge header or _meta | None |
transport_origin | Optional[str] | HTTP origin (e.g., "https://mcp.example.com") | None |
endpoint_path | Optional[str] | URL path for did:web binding (e.g., "/mcp/filesystem") | None |
config | Optional[VerifyConfig] | Verification configuration | None |
Returns:
| Type | Description |
|---|---|
VerifyResult | VerifyResult with state, trust_level, and any error details |
Example
result = await verify_server( server_did="did:web:mcp.example.com:servers:filesystem", server_badge="eyJhbGciOiJFZERTQSIsInR5cCI6IkpXVCJ9...", transport_origin="https://mcp.example.com", )
match result.state: case ServerState.VERIFIED_PRINCIPAL: print(f"Verified at level {result.trust_level}") case ServerState.DECLARED_PRINCIPAL: print("Identity claimed but not verified") case ServerState.UNVERIFIED_ORIGIN: print("No identity disclosed")
verify_server_sync ¶
verify_server_sync(
server_did: Optional[str],
server_badge: Optional[str] = None,
transport_origin: Optional[str] = None,
endpoint_path: Optional[str] = None,
config: Optional[VerifyConfig] = None,
) -> VerifyResult
Sync wrapper for verify_server.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
server_did | Optional[str] | Server DID from Capiscio-Server-DID header or _meta | required |
server_badge | Optional[str] | Server badge JWS from Capiscio-Server-Badge header or _meta | None |
transport_origin | Optional[str] | HTTP origin (e.g., "https://mcp.example.com") | None |
endpoint_path | Optional[str] | URL path for did:web binding | None |
config | Optional[VerifyConfig] | Verification configuration | None |
Returns:
| Type | Description |
|---|---|
VerifyResult | VerifyResult with state, trust_level, and any error details |
parse_http_headers ¶
Extract server identity from HTTP response headers.
RFC-007 §6.1 specifies these header names: - Capiscio-Server-DID: Server's DID - Capiscio-Server-Badge: Server's badge (JWS)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
headers | Dict[str, str] | HTTP response headers dict (case-insensitive lookup) | required |
Returns:
| Type | Description |
|---|---|
Tuple[Optional[str], Optional[str]] | Tuple of (server_did, server_badge) |
Example
did, badge = parse_http_headers(response.headers) result = await verify_server(did, badge)
parse_jsonrpc_meta ¶
Extract server identity from MCP initialize response _meta.
RFC-007 §6.2 specifies these _meta keys: - capiscio_server_did: Server's DID - capiscio_server_badge: Server's badge (JWS)
For PoP fields, use pop.PoPRequest.from_meta() and pop.PoPResponse.from_meta().
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
meta | Optional[Dict[str, Any]] | The _meta object from InitializeResult | required |
Returns:
| Type | Description |
|---|---|
Tuple[Optional[str], Optional[str]] | Tuple of (server_did, server_badge) |
PoP Module (Key Verification)¶
Proof of Possession (PoP) primitives for RFC-007 MCP server identity.
This module implements the PoP handshake for MCP server identity verification: 1. Client generates a nonce and includes it in initialize request _meta 2. Server signs the nonce with its DID key and returns signature in response _meta 3. Client verifies the signature to prove server controls the DID key
The PoP handshake upgrades server state from DECLARED_PRINCIPAL to VERIFIED_PRINCIPAL.
Usage (Client): # Generate PoP request pop_request = generate_pop_request()
# Include in initialize request _meta
meta = {
**pop_request.to_meta(),
"capiscio_server_did": server_did, # If available
}
# After receiving response, verify PoP
pop_response = PoPResponse.from_meta(response._meta)
if pop_response:
verify_pop_response(pop_request, pop_response, server_public_key)
Usage (Server): # Parse PoP request from client pop_request = PoPRequest.from_meta(request._meta)
# Create signed response
if pop_request and private_key:
pop_response = create_pop_response(pop_request, private_key, key_id)
# Include in initialize response _meta
meta = {
**identity_meta,
**pop_response.to_meta(),
}
PoPRequest dataclass ¶
PoP request sent by client in initialize request _meta.
Per RFC-007, the client generates a random nonce and includes it in the initialize request. The server must sign this nonce to prove possession of the DID key.
Attributes:
| Name | Type | Description |
|---|---|---|
client_nonce | str | Base64url-encoded random nonce (32 bytes) |
created_at | float | Timestamp when nonce was generated |
to_meta ¶
Convert to _meta format for MCP initialize request.
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dict with capiscio_pop_nonce and capiscio_pop_created_at |
from_meta classmethod ¶
Parse PoP request from MCP _meta object.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
meta | Optional[Dict[str, Any]] | The _meta dict from initialize request | required |
Returns:
| Type | Description |
|---|---|
Optional['PoPRequest'] | PoPRequest if present, None otherwise |
is_expired ¶
Check if the PoP request has expired.
PoPResponse dataclass ¶
PoP response sent by server in initialize response _meta.
Per RFC-007, the server signs the client's nonce with its DID key and returns the signature. The client verifies this signature to prove the server controls the private key for the disclosed DID.
Attributes:
| Name | Type | Description |
|---|---|---|
nonce_signature | str | JWS compact serialization of signed nonce |
signed_at | float | Timestamp when signature was created |
to_meta ¶
Convert to _meta format for MCP initialize response.
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dict with capiscio_pop_signature and capiscio_pop_signed_at |
from_meta classmethod ¶
Parse PoP response from MCP _meta object.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
meta | Optional[Dict[str, Any]] | The _meta dict from initialize response | required |
Returns:
| Type | Description |
|---|---|
Optional['PoPResponse'] | PoPResponse if present, None otherwise |
PoPError ¶
Bases: Exception
Base exception for PoP errors.
generate_pop_request ¶
Generate a new PoP request for MCP initialize.
Creates a request with a fresh nonce and current timestamp.
Returns:
| Type | Description |
|---|---|
PoPRequest | PoPRequest ready to include in _meta |
Raises:
| Type | Description |
|---|---|
PoPNonceError | If nonce generation fails |
create_pop_response ¶
create_pop_response(
request: PoPRequest,
private_key: "Ed25519PrivateKey",
key_id: str,
) -> PoPResponse
Create a PoP response by signing the client's nonce.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request | PoPRequest | The PoP request containing client nonce | required |
private_key | 'Ed25519PrivateKey' | Server's Ed25519 private key | required |
key_id | str | Key ID for JWS header (e.g., DID key reference) | required |
Returns:
| Type | Description |
|---|---|
PoPResponse | PoPResponse ready to include in _meta |
Raises:
| Type | Description |
|---|---|
PoPMissingCrypto | If cryptography package not available |
PoPSignatureError | If signing fails |
verify_pop_response ¶
verify_pop_response(
request: PoPRequest,
response: PoPResponse,
public_key: "Ed25519PublicKey",
max_age: float = POP_MAX_AGE_SECONDS,
) -> None
Verify a complete PoP response.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request | PoPRequest | The original PoP request | required |
response | PoPResponse | The PoP response from server | required |
public_key | 'Ed25519PublicKey' | Server's Ed25519 public key | required |
max_age | float | Maximum age of request in seconds | POP_MAX_AGE_SECONDS |
Raises:
| Type | Description |
|---|---|
PoPMissingCrypto | If cryptography package not available |
PoPExpiredError | If request has expired |
PoPSignatureError | If signature verification fails |
Types¶
Type definitions for capiscio-mcp.
Defines enums and dataclasses used across the package for: - RFC-006: Tool authority decisions and deny reasons - RFC-007: Server identity states and error codes
Decision ¶
Bases: str, Enum
Tool access decision result.
Per RFC-006 §6.3, every tool invocation attempt results in one of: - ALLOW: Tool execution is permitted - DENY: Tool execution is blocked with a reason
AuthLevel ¶
Bases: str, Enum
Caller authentication assurance level.
Per RFC-006 §5, every evidence log records the authentication method: - ANONYMOUS: No identity material provided - API_KEY: API key authentication (reduced assurance) - BADGE: CapiscIO Trust Badge authentication (full assurance)
DenyReason ¶
Bases: str, Enum
Reason for denying tool access.
Per RFC-006 §6.4, denial must include a specific reason code.
ServerState ¶
Bases: str, Enum
Server identity verification state.
Per RFC-007 §5.2, clients classify servers into three states:
-
VERIFIED_PRINCIPAL: Server badge verified, trust level established. The server has disclosed a DID and a valid badge signed by a trusted issuer.
-
DECLARED_PRINCIPAL: Server DID disclosed but badge missing or invalid. Identity is claimed but not cryptographically verified.
-
UNVERIFIED_ORIGIN: Server did not disclose any identity material. This is distinct from Trust Level 0 (self-signed) - UNVERIFIED_ORIGIN means NO identity was disclosed at all.
ServerErrorCode ¶
Bases: str, Enum
Server identity verification error codes.
Per RFC-007 §8, verification failures include specific error codes.
TrustLevel ¶
Bases: IntEnum
Trust levels per RFC-002 v1.4.
- LEVEL_0: Self-Signed (SS) - did:key issuer, no external validation
- LEVEL_1: Registered (REG) - Account registration with CapiscIO Registry
- LEVEL_2: Domain Validated (DV) - DNS/HTTP challenge proving domain control
- LEVEL_3: Organization Validated (OV) - DUNS/legal entity verification
- LEVEL_4: Extended Validated (EV) - Manual review + legal agreement
See: https://docs.capisc.io/rfcs/002-trust-badge/#5-trust-levels
Errors¶
Exception types for capiscio-mcp.
Provides typed exceptions for: - Guard (RFC-006) errors - Server verification (RFC-007) errors - Core connection errors
GuardError ¶
Bases: CapiscioMCPError
Raised when tool access is denied by the guard.
Per RFC-006, all denials include: - reason: Specific denial reason code - detail: Human-readable explanation - evidence_id: ID of the evidence record for audit
Example
try: result = await guarded_tool(params) except GuardError as e: logger.warning(f"Access denied: {e.reason} - {e.detail}") logger.info(f"Evidence ID: {e.evidence_id}")
GuardConfigError ¶
Bases: CapiscioMCPError
Raised when guard configuration is invalid.
This includes: - Invalid trust level values - Invalid tool name patterns - Conflicting configuration options
ServerVerifyError ¶
Bases: CapiscioMCPError
Raised when server identity verification fails.
Per RFC-007, verification can fail for various reasons including: - Invalid DID format - Badge verification failure - Origin/path binding mismatch - Trust level insufficient
Example
try: result = await verify_server(server_did, server_badge) except ServerVerifyError as e: logger.warning(f"Server verification failed: {e.error_code}") if e.state == ServerState.UNVERIFIED_ORIGIN: logger.warning("Server did not disclose identity")
CoreConnectionError ¶
Bases: CapiscioMCPError
Raised when connection to capiscio-core fails.
This includes: - Binary download failures - Process startup failures - gRPC connection failures - Health check failures
CoreVersionError ¶
Bases: CapiscioMCPError
Raised when capiscio-core version is incompatible.
capiscio-mcp requires a specific range of capiscio-core versions. This error indicates the connected core is outside that range.
Registration Module (Server Identity)¶
MCP Server Identity Registration.
This module provides functions for MCP servers to: 1. Generate Ed25519 keypairs (deriving did:key) 2. Register their DID with the CapiscIO registry
This follows the same pattern as agent identity registration in capiscio-sdk-python.
Usage
from capiscio_mcp.registration import setup_server_identity
One-step setup: generate keys + register with registry¶
result = await setup_server_identity( server_id="your-server-uuid", api_key="sk_live_...", ca_url="https://registry.capisc.io", output_dir="./keys", ) print(f"Server DID: {result['did']}") print(f"Private key saved to: {result['private_key_path']}")
Or step-by-step:¶
from capiscio_mcp.registration import generate_server_keypair, register_server_identity
Step 1: Generate keypair¶
keys = await generate_server_keypair(output_dir="./keys")
Step 2: Register with registry¶
await register_server_identity( server_id="your-server-uuid", api_key="sk_live_...", did=keys["did_key"], public_key=keys["public_key_pem"], )
RegistrationError ¶
Bases: Exception
Error during server identity registration.
KeyGenerationError ¶
Bases: Exception
Error generating keypair.
generate_server_keypair async ¶
Generate Ed25519 keypair for MCP server identity.
Uses capiscio-core's SimpleGuardService.GenerateKeyPair via gRPC. The keypair is used for PoP (Proof of Possession) verification.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key_id | str | Optional specific key ID. If empty, one is generated. | '' |
output_dir | Optional[str] | Optional directory to save private key PEM file. If provided, saves as {key_id}.pem | None |
Returns:
| Type | Description |
|---|---|
dict | dict with: - key_id: The key identifier - did_key: The derived did:key URI (e.g., did |
Raises:
| Type | Description |
|---|---|
KeyGenerationError | If key generation fails |
CoreConnectionError | If connection to capiscio-core fails |
Example
keys = await generate_server_keypair(output_dir="./keys") print(f"DID: {keys['did_key']}")
did
z6MkhaXgBZD...¶
generate_server_keypair_sync ¶
Sync wrapper for generate_server_keypair.
See generate_server_keypair() for full documentation.
register_server_identity async ¶
register_server_identity(
server_id: str,
api_key: str,
did: str,
public_key: str,
ca_url: str = "https://registry.capisc.io",
) -> dict
Register MCP server DID with the CapiscIO registry.
Uses PUT /v1/sdk/servers/{id} to update the server's DID and public key. This follows the same pattern as agent identity registration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
server_id | str | The MCP server's UUID (from dashboard creation) | required |
api_key | str | Registry API key (X-Capiscio-Registry-Key) | required |
did | str | The server's DID (e.g., did | required |
public_key | str | PEM-encoded public key | required |
ca_url | str | Registry URL (default: https://registry.capisc.io) | 'https://registry.capisc.io' |
Returns:
| Type | Description |
|---|---|
dict | dict with: - success: True if registration succeeded - message: Status message - data: Updated server object (if successful) |
Raises:
| Type | Description |
|---|---|
RegistrationError | If registration fails |
Example
result = await register_server_identity( server_id="550e8400-e29b-41d4-a716-446655440000", api_key="sk_live_abc123...", did="didz6MkhaXgBZD...", public_key="-----BEGIN PUBLIC KEY-----...", )
register_server_identity_sync ¶
register_server_identity_sync(
server_id: str,
api_key: str,
did: str,
public_key: str,
ca_url: str = "https://registry.capisc.io",
) -> dict
Sync wrapper for register_server_identity.
See register_server_identity() for full documentation.
setup_server_identity async ¶
setup_server_identity(
server_id: str,
api_key: str,
ca_url: str = "https://registry.capisc.io",
output_dir: Optional[str] = None,
key_id: str = "",
) -> dict
Generate keypair and register server identity in one call.
This is the recommended way to set up MCP server identity: 1. Generates Ed25519 keypair via capiscio-core 2. Registers the DID with the CapiscIO registry 3. Optionally saves the private key to disk
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
server_id | str | The MCP server's UUID (from dashboard creation) | required |
api_key | str | Registry API key (X-Capiscio-Registry-Key) | required |
ca_url | str | Registry URL (default: https://registry.capisc.io) | 'https://registry.capisc.io' |
output_dir | Optional[str] | Optional directory to save private key PEM file | None |
key_id | str | Optional specific key ID. If empty, one is generated. | '' |
Returns:
| Type | Description |
|---|---|
dict | dict with: - did: The server's DID (did |
Raises:
| Type | Description |
|---|---|
KeyGenerationError | If key generation fails |
RegistrationError | If registry registration fails |
CoreConnectionError | If connection to capiscio-core fails |
setup_server_identity_sync ¶
setup_server_identity_sync(
server_id: str,
api_key: str,
ca_url: str = "https://registry.capisc.io",
output_dir: Optional[str] = None,
key_id: str = "",
) -> dict
Sync wrapper for setup_server_identity.
See setup_server_identity() for full documentation.
MCP SDK Integration¶
MCP SDK Integration — requires pip install capiscio-mcp[mcp]
Provides two separate integration classes: 1. Server-side: CapiscioMCPServer (guard tools, disclose identity, PoP signing) 2. Client-side: CapiscioMCPClient (verify server identity, PoP verification)
Usage (Server): from capiscio_mcp.integrations.mcp import CapiscioMCPServer
server = CapiscioMCPServer(
name="filesystem",
did="did:web:mcp.example.com:servers:filesystem",
badge="eyJhbGc...",
private_key_path="/path/to/key.pem", # For PoP signing
)
@server.tool(min_trust_level=2)
async def read_file(path: str) -> str:
with open(path) as f:
return f.read()
# Run the server
server.run()
Usage (Client): from capiscio_mcp.integrations.mcp import CapiscioMCPClient
async with CapiscioMCPClient(
server_url="https://mcp.example.com",
min_trust_level=2,
require_pop=True, # Require PoP verification
) as client:
result = await client.call_tool("read_file", {"path": "/data/file.txt"})
CapiscioMCPServer ¶
MCP Server with CapiscIO identity disclosure, PoP signing, and tool guarding.
This class wraps FastMCP to: 1. Automatically inject identity into initialize response _meta 2. Sign PoP challenges to prove key ownership (RFC-007) 3. Guard registered tools with @guard decorator for trust enforcement
Attributes:
| Name | Type | Description |
|---|---|---|
name | Server name | |
did | Server DID (did:web:... or did | |
badge | Server trust badge JWS (optional but recommended) | |
default_min_trust_level | Default minimum trust level for tools | |
pop_enabled | bool | Whether PoP signing is available |
Example
server = CapiscioMCPServer( name="filesystem", did="did:web:mcp.example.com:servers:filesystem", badge=os.environ.get("SERVER_BADGE"), private_key_path="/path/to/server.key.pem", )
@server.tool(min_trust_level=2) async def read_file(path: str) -> str: with open(path) as f: return f.read()
Run the server¶
server.run()
identity_meta property ¶
Get the identity metadata for initialize response.
__init__ ¶
__init__(
name: str,
did: str,
badge: Optional[str] = None,
default_min_trust_level: int = 0,
version: str = "1.0.0",
private_key: Optional["Ed25519PrivateKey"] = None,
private_key_path: Optional[str] = None,
private_key_pem: Optional[Union[str, bytes]] = None,
key_id: Optional[str] = None,
)
Initialize CapiscIO MCP Server.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name | str | Server name (shown to clients) | required |
did | str | Server DID for identity disclosure | required |
badge | Optional[str] | Server badge JWS for identity verification | None |
default_min_trust_level | int | Default minimum trust level for tools | 0 |
version | str | Server version string | '1.0.0' |
private_key | Optional['Ed25519PrivateKey'] | Ed25519 private key for PoP signing (optional) | None |
private_key_path | Optional[str] | Path to PEM file containing private key (optional) | None |
private_key_pem | Optional[Union[str, bytes]] | PEM-encoded private key string/bytes (optional) | None |
key_id | Optional[str] | Key ID for JWS header (defaults to DID#keys-1) | None |
create_initialize_response_meta ¶
Create the _meta object for initialize response.
This method should be called when building the initialize response. It includes: 1. Server identity (DID, badge) 2. PoP response (if client sent PoP request and we have a private key)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request_meta | Optional[Dict[str, Any]] | The _meta from the initialize request (for PoP) | None |
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dict to include as _meta in initialize response |
Example
In your initialize handler¶
def handle_initialize(request): response_meta = server.create_initialize_response_meta( request_meta=request.params.get("_meta") ) return InitializeResult( capabilities=..., _meta=response_meta, )
tool ¶
tool(
name: Optional[str] = None,
description: Optional[str] = None,
min_trust_level: Optional[int] = None,
config: Optional[GuardConfig] = None,
) -> Callable[
[Callable[..., Coroutine[Any, Any, T]]],
Callable[..., Coroutine[Any, Any, T]],
]
Register a tool with CapiscIO guard.
This decorator: 1. Registers the function as an MCP tool via FastMCP 2. Wraps it with @guard for access control based on caller trust level
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name | Optional[str] | Tool name (default: function name) | None |
description | Optional[str] | Tool description | None |
min_trust_level | Optional[int] | Minimum trust level (overrides default) | None |
config | Optional[GuardConfig] | Full guard configuration | None |
Returns:
| Type | Description |
|---|---|
Callable[[Callable[..., Coroutine[Any, Any, T]]], Callable[..., Coroutine[Any, Any, T]]] | Decorator function |
Example
@server.tool(min_trust_level=2) async def execute_query(sql: str) -> list[dict]: ...
run ¶
Run the server with the specified transport.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
transport | str | Transport type - "stdio" (default) or "streamable-http" | 'stdio' |
Example
server.run() # stdio transport server.run(transport="streamable-http") # HTTP transport
run_sse ¶
Run the server over SSE transport.
Deprecated: Use run(transport="sse") instead. SSE is deprecated in favor of streamable-http.
CapiscioMCPClient ¶
MCP Client with automatic server identity and PoP verification.
This class wraps MCP client functionality to: 1. Generate PoP request (nonce) for initialize request 2. Verify server identity and PoP response on connection 3. Enforce trust level requirements 4. Include caller credentials in tool requests
Attributes:
| Name | Type | Description |
|---|---|---|
server_url | URL of the MCP server | |
min_trust_level | Minimum required trust level | |
fail_on_unverified | If True, raise on unverified servers | |
require_pop | If True, require PoP verification (did:key servers) | |
pop_verified | bool | Whether PoP verification succeeded |
Example
async with CapiscioMCPClient( server_url="https://mcp.example.com", min_trust_level=2, require_pop=True, badge="eyJhbGc...", # Your client badge ) as client: # Server identity and PoP already verified print(f"Trusted at level {client.server_trust_level}") print(f"PoP verified: {client.pop_verified}")
result = await client.call_tool("read_file", {"path": "/data/file.txt"})
For stdio transport (subprocess server): async with CapiscioMCPClient( command="python", args=["my_mcp_server.py"], min_trust_level=1, ) as client: result = await client.call_tool("my_tool", {"arg": "value"})
server_state property ¶
Server verification state after connection.
__init__ ¶
__init__(
server_url: Optional[str] = None,
command: Optional[str] = None,
args: Optional[List[str]] = None,
min_trust_level: int = 0,
fail_on_unverified: bool = True,
require_pop: bool = False,
verify_config: Optional[VerifyConfig] = None,
badge: Optional[str] = None,
api_key: Optional[str] = None,
)
Initialize CapiscIO MCP Client.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
server_url | Optional[str] | URL of the MCP server (for HTTP transport) | None |
command | Optional[str] | Command to run server (for stdio transport) | None |
args | Optional[List[str]] | Arguments for command (for stdio transport) | None |
min_trust_level | int | Minimum required server trust level | 0 |
fail_on_unverified | bool | If True, raise when server doesn't disclose identity | True |
require_pop | bool | If True, require PoP verification for did:key servers | False |
verify_config | Optional[VerifyConfig] | Full verification configuration | None |
badge | Optional[str] | Client badge for authentication (recommended) | None |
api_key | Optional[str] | Client API key for authentication (alternative) | None |
Raises:
| Type | Description |
|---|---|
ValueError | If neither server_url nor command is provided |
create_initialize_request_meta ¶
Create the _meta object for initialize request.
This should be called when building the initialize request. It generates a PoP nonce to be signed by the server.
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dict to include as _meta in initialize request |
Example
In your client code¶
meta = client.create_initialize_request_meta() result = await session.initialize( client_info=ClientInfo(...), _meta=meta, )
verify_initialize_response ¶
verify_initialize_response(
response_meta: Optional[Dict[str, Any]],
server_public_key: Optional["Ed25519PublicKey"] = None,
) -> bool
Verify the initialize response including PoP.
This should be called after receiving the initialize response. It extracts the PoP signature and verifies it.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
response_meta | Optional[Dict[str, Any]] | The _meta from initialize response | required |
server_public_key | Optional['Ed25519PublicKey'] | Server's public key for PoP verification (if None, will try to extract from did:key) | None |
Returns:
| Type | Description |
|---|---|
bool | True if PoP verification succeeded, False otherwise |
Raises:
| Type | Description |
|---|---|
PoPSignatureError | If PoP verification fails and require_pop=True |
__aenter__ async ¶
Async context manager entry.
Connects to server and verifies identity.
__aexit__ async ¶
Async context manager exit.
connect async ¶
Connect to MCP server.
For stdio transport, spawns the server process. For HTTP transport, connects to the server URL (not yet implemented).
Note
Server identity verification (min_trust_level, fail_on_unverified) is not yet functional due to MCP SDK limitations. The SDK does not currently expose _meta from the initialize response, which is needed to extract server DID and badge. Server-side trust enforcement works fully.
Raises:
| Type | Description |
|---|---|
NotImplementedError | If HTTP transport is requested (not yet supported) |
call_tool async ¶
Call a tool on the connected server.
Automatically includes client credentials in the request.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name | str | Tool name | required |
arguments | Optional[Dict[str, Any]] | Tool arguments | None |
Returns:
| Type | Description |
|---|---|
Any | Tool result from the server |
Raises:
| Type | Description |
|---|---|
RuntimeError | If not connected |
list_tools async ¶
List available tools on the server.
Returns:
| Type | Description |
|---|---|
List[Dict[str, Any]] | List of tool definitions |