Skip to content

ccproxy.api.routes.mcp

ccproxy.api.routes.mcp

MCP (Model Context Protocol) server for CCProxy API Server.

Provides MCP server functionality including permission checking tools.

PermissionCheckRequest

Bases: BaseModel

Request model for permission checking.

check_permission async

check_permission(request, settings)

Check permissions for a tool call.

This implements the same security logic as the CLI permission tool, checking for dangerous patterns and restricted tools.

Source code in ccproxy/api/routes/mcp.py
async def check_permission(
    request: PermissionCheckRequest,
    settings: SettingsDep,
) -> (
    PermissionToolAllowResponse
    | PermissionToolDenyResponse
    | PermissionToolPendingResponse
):
    """Check permissions for a tool call.

    This implements the same security logic as the CLI permission tool,
    checking for dangerous patterns and restricted tools.
    """
    logger.info(
        "permission_check",
        tool_name=request.tool_name,
        retry=request.permission_id is not None,
    )

    permission_service = get_permission_service()

    if request.permission_id:
        status = await permission_service.get_status(request.permission_id)

        if status == PermissionStatus.ALLOWED:
            return PermissionToolAllowResponse(updated_input=request.input)

        elif status == PermissionStatus.DENIED:
            return PermissionToolDenyResponse(message="User denied the operation")

        elif status == PermissionStatus.EXPIRED:
            return PermissionToolDenyResponse(message="Permission request expired")

    logger.info(
        "permission_requires_authorization",
        tool_name=request.tool_name,
    )

    permission_id = await permission_service.request_permission(
        tool_name=request.tool_name,
        input=request.input,
    )

    # Wait for permission to be resolved
    try:
        final_status = await permission_service.wait_for_permission(
            permission_id,
            timeout_seconds=settings.security.confirmation_timeout_seconds,
        )

        if final_status == PermissionStatus.ALLOWED:
            logger.info(
                "permission_allowed_after_authorization",
                tool_name=request.tool_name,
                permission_id=permission_id,
            )
            return PermissionToolAllowResponse(updated_input=request.input)
        else:
            logger.info(
                "permission_denied_after_authorization",
                tool_name=request.tool_name,
                permission_id=permission_id,
                status=final_status.value,
            )
            return PermissionToolDenyResponse(
                message=f"User denied the operation (status: {final_status.value})"
            )

    except TimeoutError:
        logger.warning(
            "permission_authorization_timeout",
            tool_name=request.tool_name,
            permission_id=permission_id,
            timeout_seconds=settings.security.confirmation_timeout_seconds,
        )
        return PermissionToolDenyResponse(message="Permission request timed out")

setup_mcp

setup_mcp(app)

Set up MCP server on the given FastAPI app.

Parameters:

Name Type Description Default
app FastAPI

The FastAPI application to mount MCP on

required
Source code in ccproxy/api/routes/mcp.py
def setup_mcp(app: FastAPI) -> None:
    """Set up MCP server on the given FastAPI app.

    Args:
        app: The FastAPI application to mount MCP on
    """
    # Minimal MCP sub-app without middleware or docs
    mcp_app = FastAPI(
        title="CCProxy MCP Server",
        description="MCP server for Claude Code permission checking",
        openapi_url=None,
        docs_url=None,
        redoc_url=None,
    )

    @mcp_app.post(
        "/permission/check",
        operation_id="check_permission",
        summary="Check permissions for a tool call",
        description="Validates whether a tool call should be allowed based on security rules",
        response_model=PermissionToolAllowResponse
        | PermissionToolDenyResponse
        | PermissionToolPendingResponse,
        tags=["mcp-tools"],
    )
    async def permission_endpoint(
        request: PermissionCheckRequest,
        settings: SettingsDep,
    ) -> (
        PermissionToolAllowResponse
        | PermissionToolDenyResponse
        | PermissionToolPendingResponse
    ):
        """Check permissions for a tool call."""
        return await check_permission(request, settings)

    mcp = FastApiMCP(
        mcp_app,
        name="CCProxy MCP Server",
        description="MCP server for Claude Code permission checking",
        include_operations=["check_permission"],
    )

    mcp.mount(app, mount_path="/mcp")