invoke()
Executes an action on an MCP server with cryptographic verification.
client.invoke(
mcp: str,
action: str,
intent_token: IntentToken,
params: dict = None,
merkle_proof: list = None,
user_email: str = None
) -> MCPInvocationResultParameters
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
| mcp | str | Yes | - | MCP server name (e.g., "analytics-mcp") |
| action | str | Yes | - | Action/tool to execute (must be in plan) |
| intent_token | IntentToken | Yes | - | Token from get_intent_token() |
| params | dict | No | Action parameters | |
| merkle_proof | list | No | Auto-generated | Optional Merkle proof |
| user_email | str | No | None | Optional user email |
Flow
- SDK generates Merkle proof for this action from plan
- SDK → ArmorIQ Proxy POST /invoke with CSRG headers:
- X-API-Key: API key for authentication
- X-CSRG-Path: Path in plan (e.g., /steps/[0]/action)
- X-CSRG-Value-Digest: SHA256 hash of action value
- X-CSRG-Proof: JSON Merkle proof array
- Proxy performs IAP Step Verification:
- Validates Merkle proof against plan_hash
- Verifies CSRG path matches plan structure
- Checks value digest matches action
- Verifies Ed25519 signature on token
- If verification passes, proxy routes to MCP server
- MCP executes action and returns result
Returns
dict containing:
{
"success": bool, # Whether action succeeded
"data": any, # Response data from MCP
"error": str, # Error message (if failed)
"execution_time_ms": int, # Execution duration
"mcp": str, # MCP that executed
"action": str # Action that ran
}Raises
- VerificationError: If IAP Step Verification fails
- TokenExpiredError: If token has expired
- MCPError: If MCP execution fails
- NetworkError: If request fails
Example
# Basic invocation
result = client.invoke(
mcp="analytics-mcp",
action="analyze",
intent_token=token,
params={"data": [1, 2, 3, 4, 5], "metrics": ["mean", "std"]}
)
if result["success"]:
print(f"Results: {result['data']}")
print(f"Took: {result['execution_time_ms']}ms")
else:
print(f"Error: {result['error']}")
# With error handling
try:
result = client.invoke("data-mcp", "fetch_data", token, {"source": "db"})
if result["success"]:
data = result["data"]
else:
logger.error(f"MCP error: {result['error']}")
except TokenExpiredError:
# Get fresh token
token = client.get_intent_token(plan)["token"]
result = client.invoke("data-mcp", "fetch_data", token, {"source": "db"})
except VerificationError as e:
# Action not in plan
logger.error(f"Verification failed: {e}")
# Need to recreate plan with correct actions
# Custom timeout
result = client.invoke(
"analytics-mcp",
"long_analysis",
token,
{"dataset": "large"},
timeout=120 # 2 minutes
)