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 - One-line server identity setup via MCPServerIdentity.connect()
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 ("Let's Encrypt" style โ recommended): from capiscio_mcp import MCPServerIdentity from capiscio_mcp.integrations.mcp import CapiscioMCPServer
identity = await MCPServerIdentity.connect(
server_id=os.environ["CAPISCIO_SERVER_ID"],
api_key=os.environ["CAPISCIO_API_KEY"],
)
server = CapiscioMCPServer(identity=identity)
@server.tool(min_trust_level=2)
async def read_file(path: str) -> str:
...
server.run()
Quickstart (@guard decorator): 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, manual): 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 integration classes:
- Server-side: :class:
CapiscioMCPServerโ guard tools, disclose identity via_metain the initialize response, and sign PoP challenges. - Client-side: :class:
CapiscioMCPClientโ verify server identity extracted from_metaon connection, enforcemin_trust_level.
Usage (Server)::
from capiscio_mcp.integrations.mcp import CapiscioMCPServer
from capiscio_mcp import MCPServerIdentity
identity = await MCPServerIdentity.connect(
server_id=os.environ["CAPISCIO_SERVER_ID"],
api_key=os.environ["CAPISCIO_API_KEY"],
)
server = CapiscioMCPServer(identity=identity)
@server.tool(min_trust_level=2)
async def read_file(path: str) -> str:
with open(path) as f:
return f.read()
server.run()
Usage (Client)::
from capiscio_mcp.integrations.mcp import CapiscioMCPClient
async with CapiscioMCPClient(
command="python server/main.py",
badge="eyJhbGc...",
min_trust_level=1,
fail_on_unverified=True,
) as client:
result = await client.call_tool("list_files", {"directory": "/tmp"})
CapiscioMCPServer ¶
MCP Server with CapiscIO identity disclosure, PoP signing, and tool guarding.
This class wraps FastMCP to:
- Automatically inject server identity (DID, badge) into the
_metafield of MCPinitializeresponses (RFC-007 ยง6.2). - Sign PoP challenges from clients to prove key ownership (RFC-007).
- Guard registered tools with trust-level requirements (RFC-006).
The _meta injection works by patching mcp.server.session.ServerSession._received_request once (globally, idempotently) at server start-time. A per-invocation contextvar carries the meta dict so that only this server's responses are affected.
Example::
from capiscio_mcp import MCPServerIdentity
from capiscio_mcp.integrations.mcp import CapiscioMCPServer
identity = await MCPServerIdentity.connect(
server_id=os.environ["CAPISCIO_SERVER_ID"],
api_key=os.environ["CAPISCIO_API_KEY"],
)
server = CapiscioMCPServer(identity=identity)
@server.tool(min_trust_level=2)
async def read_file(path: str) -> str:
with open(path) as f:
return f.read()
server.run()
You can also supply credentials directly (without MCPServerIdentity)::
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",
)
identity_meta property ¶
The identity _meta dict (DID + badge) used in initialize responses.
__init__ ¶
__init__(
name: Optional[str] = None,
did: Optional[str] = None,
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,
identity: Optional[Any] = None,
) -> None
Initialize CapiscioMCPServer.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name | Optional[str] | Server name shown to clients. | None |
did | Optional[str] | Server DID for identity disclosure. | None |
badge | Optional[str] | Server badge JWS for client 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'] |
| 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 | None |
identity | Optional[Any] | :class: | None |
create_initialize_response_meta ¶
Build the _meta dict for an initialize response (RFC-007 ยง6.2).
Includes server identity (DID, badge) and, if the client sent a PoP nonce and a private key is loaded, a PoP signature.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request_meta | Optional[Dict[str, Any]] | The | None |
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dict to be included as |
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 trust-level guarding.
Wraps the function with :func:~capiscio_mcp.guard.guard for access control based on caller trust level, then registers it with FastMCP.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name | Optional[str] | Tool name (defaults to function name). | None |
description | Optional[str] | Tool description (defaults to docstring). | None |
min_trust_level | Optional[int] | Minimum trust level (overrides server default). | None |
config | Optional[GuardConfig] | Full :class: | None |
Example::
@server.tool(min_trust_level=2)
async def execute_query(sql: str) -> list[dict]:
...
run ¶
Run the server with _meta identity injection enabled.
Sets _capiscio_meta_ctx so that the patched ServerSession injects the CapiscIO identity into every initialize response for the lifetime of this call.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
transport | str |
| 'stdio' |
run_sse ¶
Run over SSE transport (deprecated โ use run(transport='streamable-http')).
CapiscioMCPClient ¶
MCP Client with automatic server identity and PoP verification.
On connection this client:
- Calls
session.initialize()to get the MCPInitializeResult. - Extracts
capiscio_server_didandcapiscio_server_badgefromresult.meta(the_metadict in the JSON response). - Calls :func:
~capiscio_mcp.server.verify_serverwith those values. - Enforces
min_trust_levelandfail_on_unverifiedconstraints.
Example::
async with CapiscioMCPClient(
command="python server/main.py",
badge=agent.badge,
min_trust_level=1,
fail_on_unverified=True,
) as client:
print(f"Server verified at level {client.server_trust_level}")
files = await client.call_tool("list_files", {"directory": "/tmp"})
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,
env: Optional[Dict[str, str]] = None,
) -> None
Initialize CapiscioMCPClient.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
server_url | Optional[str] | MCP server URL (HTTP transport โ not yet implemented). | None |
command | Optional[str] | Command to launch server process (stdio transport). | None |
args | Optional[List[str]] | Arguments for the server command. | None |
min_trust_level | int | Minimum required server trust level. | 0 |
fail_on_unverified | bool | If | True |
require_pop | bool | If | False |
verify_config | Optional[VerifyConfig] | Full :class: | None |
badge | Optional[str] | Client badge JWS for authentication. | None |
api_key | Optional[str] | Client API key for authentication (alternative to badge). | None |
env | Optional[Dict[str, str]] | Additional environment variables to pass to the server subprocess (stdio transport only). All | None |
Raises:
| Type | Description |
|---|---|
ValueError | If neither |
ImportError | If the |
create_initialize_request_meta ¶
Create the _meta dict for the initialize request (PoP nonce).
Returns:
| Type | Description |
|---|---|
Dict[str, Any] | Dict to include as |
verify_initialize_response ¶
verify_initialize_response(
response_meta: Optional[Dict[str, Any]],
server_public_key: Optional["Ed25519PublicKey"] = None,
) -> bool
Verify the initialize response including optional PoP.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
response_meta | Optional[Dict[str, Any]] | The | required |
server_public_key | Optional['Ed25519PublicKey'] | Server public key for PoP (auto-extracted from | None |
Returns:
| Type | Description |
|---|---|
bool |
|
Raises:
| Type | Description |
|---|---|
| class: |
connect async ¶
Connect to the MCP server and verify its identity.
For stdio transport, spawns the server process. Extracts _meta from the InitializeResult and calls :func:~capiscio_mcp.server.verify_server to enforce min_trust_level and fail_on_unverified constraints.
Raises:
| Type | Description |
|---|---|
| class: | |
| exc: |
call_tool async ¶
Call a tool on the connected server, forwarding client credentials.
For stdio transport the caller's badge (and/or API key) is forwarded in the JSON-RPC _meta of the tools/call request under the keys capiscio_caller_badge / capiscio_caller_api_key. The server-side :func:_install_credential_extraction wrapper picks these up and sets the _current_credential contextvar before the guarded tool runs.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name | str | Tool name. | required |
arguments | Optional[Dict[str, Any]] | Tool arguments dict. | None |
Returns:
| Type | Description |
|---|---|
Any | Tool result from the server. |
Raises:
| Type | Description |
|---|---|
| exc: |
list_tools async ¶
List tools available on the connected server.
Returns:
| Type | Description |
|---|---|
List[Dict[str, Any]] | List of |
Raises:
| Type | Description |
|---|---|
| exc: |