MCP Integration for Python

Integrate ASCEND governance into your MCP (Model Context Protocol) server tools using decorators.

Overview

The ASCEND Python SDK provides seamless integration with MCP servers through the mcp_governance decorator. This enables automatic authorization checks before any tool execution, ensuring your AI agents operate within policy boundaries.

Installation

pip install ascend-ai-sdk

Quick Start

from ascend import AscendClient
from ascend.mcp import mcp_governance

# Initialize ASCEND client
client = AscendClient(
    api_key="owkai_your_api_key",
    agent_id="mcp-server-001",
    agent_name="My MCP Server"
)

# Apply governance to MCP tool
@mcp_server.tool()
@mcp_governance(client, action_type="database.query", resource="production_db")
async def query_database(sql: str) -> dict:
    """Execute a database query with ASCEND governance."""
    return await db.execute(sql)

Decorator Reference

mcp_governance()

The primary decorator for adding governance to MCP tools.

from ascend.mcp import mcp_governance

@mcp_governance(
    client: AscendClient,
    action_type: str,
    resource: str,
    config: MCPGovernanceConfig = None,
    risk_level: str = None,
    metadata: Dict[str, Any] = None,
    require_human_approval: bool = False
)

Parameters

Parameter	Type	Required	Description
client	AscendClient	Yes	Initialized ASCEND client
action_type	str	Yes	Action type (e.g., "database.query")
resource	str	Yes	Resource being accessed
config	MCPGovernanceConfig	No	Custom configuration
risk_level	str	No	Override risk level ("low", "medium", "high", "critical")
metadata	Dict	No	Additional metadata
require_human_approval	bool	No	Force human approval

Example

@mcp_server.tool()
@mcp_governance(
    client,
    action_type="database.write",
    resource="production_db",
    risk_level="high",
    require_human_approval=True,
    metadata={"category": "data_modification"}
)
async def delete_records(table: str, where_clause: str) -> dict:
    """Delete records from database with mandatory approval."""
    return await db.execute(f"DELETE FROM {table} WHERE {where_clause}")

---

require_governance()

Simplified alias for common use cases.

from ascend.mcp import require_governance

@mcp_server.tool()
@require_governance(client, "file.read", "/var/log")
async def read_logs(filename: str) -> str:
    with open(f"/var/log/{filename}") as f:
        return f.read()

---

high_risk_action()

Decorator for actions that always require human approval.

from ascend.mcp import high_risk_action

@mcp_server.tool()
@high_risk_action(client, "system.shutdown", "server-001")
async def shutdown_server() -> dict:
    """Shutdown server - always requires approval."""
    return await system.shutdown()

---

MCPGovernanceConfig

Fine-tune governance behavior with configuration options.

from ascend.mcp import MCPGovernanceConfig, mcp_governance

config = MCPGovernanceConfig(
    # Approval handling
    wait_for_approval=True,
    approval_timeout_seconds=300,
    approval_poll_interval_seconds=5,

    # Context enrichment
    include_tool_name=True,
    include_arguments=True,
    include_caller_info=True,

    # Error handling
    raise_on_denial=True,
    log_all_decisions=True,

    # Custom callbacks
    on_approval_required=my_approval_callback,
    on_denied=my_denial_callback,
    on_allowed=my_allowed_callback,
    on_timeout=my_timeout_callback
)

@mcp_server.tool()
@mcp_governance(client, "api.call", "external_api", config=config)
async def call_external_api(endpoint: str, data: dict) -> dict:
    return await api.post(endpoint, data)

Configuration Options

Option	Type	Default	Description
wait_for_approval	bool	True	Wait for approval if action is pending
approval_timeout_seconds	int	300	Max wait time for approval (5 min)
approval_poll_interval_seconds	int	5	Polling interval for status checks
include_tool_name	bool	True	Include tool name in context
include_arguments	bool	True	Include sanitized arguments in context
include_caller_info	bool	True	Include caller information
raise_on_denial	bool	True	Raise exception on denial
log_all_decisions	bool	True	Log all authorization decisions

Callback Functions

async def on_approval_required(decision, tool_name):
    """Called when action requires human approval."""
    await notify_slack(
        f"Action '{tool_name}' requires approval",
        approval_id=decision.approval_request_id
    )

async def on_denied(decision, tool_name):
    """Called when action is denied."""
    await log_security_event(
        event="action_denied",
        tool=tool_name,
        reason=decision.reason
    )

async def on_allowed(decision, tool_name):
    """Called when action is allowed."""
    await metrics.increment("actions_allowed", tags={"tool": tool_name})

async def on_timeout(decision, tool_name):
    """Called when approval times out."""
    await alert_ops(f"Approval timeout for {tool_name}")

---

MCPGovernanceMiddleware

Apply governance to multiple tools using a middleware pattern.

from ascend.mcp import MCPGovernanceMiddleware

# Create middleware instance
middleware = MCPGovernanceMiddleware(client, default_config=config)

# Apply to multiple tools
@mcp_server.tool()
@middleware.govern("database.query", "production_db")
async def query(sql: str) -> dict:
    return await db.execute(sql)

@mcp_server.tool()
@middleware.govern("file.read", "/var/log")
async def read_file(path: str) -> str:
    return await read_async(path)

@mcp_server.tool()
@middleware.govern("api.call", "external_api")
async def api_request(url: str, method: str = "GET") -> dict:
    return await http.request(url, method)

# List governed tools
print(f"Governed tools: {middleware.governed_tools}")
# Output: ['query', 'read_file', 'api_request']

---

Complete MCP Server Example

"""
Complete MCP server with ASCEND governance
"""
import asyncio
from mcp import Server, Tool
from ascend import AscendClient, FailMode
from ascend.mcp import (
    mcp_governance,
    high_risk_action,
    MCPGovernanceConfig,
    MCPGovernanceMiddleware
)

# Initialize ASCEND client
ascend = AscendClient(
    api_key="owkai_prod_xxxxxxxxxxxx",
    agent_id="data-assistant-mcp",
    agent_name="Data Assistant MCP Server",
    environment="production",
    fail_mode=FailMode.CLOSED
)

# Register the MCP server as an agent
ascend.register(
    agent_type="mcp_server",
    capabilities=[
        "database.query",
        "database.write",
        "file.read",
        "file.write",
        "api.call"
    ],
    allowed_resources=[
        "production_db",
        "analytics_db",
        "/var/data",
        "internal_api"
    ]
)

# Create MCP server
server = Server("data-assistant")

# Custom configuration with notifications
config = MCPGovernanceConfig(
    wait_for_approval=True,
    approval_timeout_seconds=600,
    on_approval_required=lambda d, t: print(f"Approval needed for {t}"),
    on_denied=lambda d, t: print(f"DENIED: {t} - {d.reason}")
)


# Low-risk tool: database read
@server.tool()
@mcp_governance(ascend, "database.query", "analytics_db")
async def analyze_data(query: str) -> dict:
    """
    Run analytics query on the analytics database.

    Args:
        query: SQL query to execute (SELECT only)
    """
    if not query.strip().upper().startswith("SELECT"):
        raise ValueError("Only SELECT queries allowed")

    result = await db.execute(query)
    return {"rows": result, "count": len(result)}


# Medium-risk tool: file operations
@server.tool()
@mcp_governance(
    ascend,
    action_type="file.read",
    resource="/var/data",
    risk_level="medium"
)
async def read_data_file(filename: str) -> str:
    """
    Read a data file from the data directory.

    Args:
        filename: Name of file to read (within /var/data)
    """
    import os
    safe_path = os.path.join("/var/data", os.path.basename(filename))
    with open(safe_path) as f:
        return f.read()


# High-risk tool: database write
@server.tool()
@mcp_governance(
    ascend,
    action_type="database.write",
    resource="production_db",
    risk_level="high",
    config=config
)
async def update_records(table: str, updates: dict, where: str) -> dict:
    """
    Update records in production database.

    Args:
        table: Table name
        updates: Column-value pairs to update
        where: WHERE clause for the update
    """
    set_clause = ", ".join(f"{k} = %s" for k in updates.keys())
    query = f"UPDATE {table} SET {set_clause} WHERE {where}"

    result = await db.execute(query, list(updates.values()))
    return {"affected_rows": result.rowcount}


# Critical-risk tool: always requires human approval
@server.tool()
@high_risk_action(ascend, "database.delete", "production_db")
async def delete_records(table: str, where: str) -> dict:
    """
    Delete records from production database.
    CRITICAL: Always requires human approval.

    Args:
        table: Table name
        where: WHERE clause for deletion
    """
    query = f"DELETE FROM {table} WHERE {where}"
    result = await db.execute(query)
    return {"deleted_rows": result.rowcount}


# External API call with metadata
@server.tool()
@mcp_governance(
    ascend,
    action_type="api.call",
    resource="internal_api",
    metadata={"api_type": "internal", "requires_auth": True}
)
async def fetch_user_data(user_id: str) -> dict:
    """
    Fetch user data from internal API.

    Args:
        user_id: User ID to look up
    """
    response = await http.get(f"https://api.internal/users/{user_id}")
    return response.json()


# Run the server
if __name__ == "__main__":
    asyncio.run(server.run())

---

Error Handling

The MCP integration provides comprehensive error handling:

from ascend import AuthorizationError, TimeoutError, CircuitBreakerOpen

@server.tool()
@mcp_governance(ascend, "database.query", "production_db")
async def query_with_handling(sql: str) -> dict:
    """Query with explicit error handling."""
    try:
        return await db.execute(sql)

    except AuthorizationError as e:
        # Action was denied
        return {
            "error": "Authorization denied",
            "reason": e.message,
            "policy_violations": e.policy_violations,
            "risk_score": e.risk_score
        }

    except TimeoutError as e:
        # Approval timed out
        return {
            "error": "Approval timeout",
            "timeout_seconds": e.timeout_seconds,
            "message": "Action requires approval but timed out waiting"
        }

    except CircuitBreakerOpen as e:
        # ASCEND service unavailable
        return {
            "error": "Governance service unavailable",
            "recovery_time": e.recovery_time,
            "message": "Please try again later"
        }

---

Best Practices

1. Use Specific Action Types

# Good: Specific action type
@mcp_governance(client, "database.query", "production_db")

# Bad: Generic action type
@mcp_governance(client, "action", "db")

2. Include Meaningful Context

@mcp_governance(
    client,
    "api.call",
    "payment_api",
    metadata={
        "operation": "refund",
        "category": "financial",
        "requires_audit": True
    }
)

3. Use Appropriate Risk Levels

# Read operations: default (low)
@mcp_governance(client, "database.query", "db")

# Write operations: medium to high
@mcp_governance(client, "database.write", "db", risk_level="high")

# Destructive operations: high with approval
@high_risk_action(client, "database.delete", "db")

4. Handle Pending Approvals Gracefully

config = MCPGovernanceConfig(
    wait_for_approval=True,
    approval_timeout_seconds=300,
    on_approval_required=async lambda d, t: await notify_user(
        f"Your request requires approval. Tracking ID: {d.approval_request_id}"
    )
)

5. Configure Fail Mode Appropriately

# Production: fail-closed (secure)
prod_client = AscendClient(
    api_key="owkai_prod_xxx",
    fail_mode=FailMode.CLOSED
)

# Development: fail-open (for testing)
dev_client = AscendClient(
    api_key="owkai_dev_xxx",
    fail_mode=FailMode.OPEN
)

---

See Also

• Python SDK Quickstart
• Python SDK Reference
• Code Examples