Function register_client_session_api_handlers

python/copilot/generated/rpc.py:26937–27053

Register client-session request handlers on a JSON-RPC connection.

(
    client: "JsonRpcClient",
    get_handlers: Callable[[str], ClientSessionApiHandlers],
)

    canvas: CanvasHandler | None = None

def register_client_session_api_handlers(
    client: "JsonRpcClient",
    get_handlers: Callable[[str], ClientSessionApiHandlers],
) -> None:
    """Register client-session request handlers on a JSON-RPC connection."""
    async def handle_provider_token_get_token(params: dict) -> dict | None:
        request = ProviderTokenAcquireRequest.from_dict(params)
        handler = get_handlers(request.session_id).provider_token
        if handler is None: raise RuntimeError(f"No provider_token handler registered for session: {request.session_id}")
        result = await handler.get_token(request)
        return result.to_dict()
    client.set_request_handler("providerToken.getToken", handle_provider_token_get_token)
    async def handle_session_fs_read_file(params: dict) -> dict | None:
        request = SessionFSReadFileRequest.from_dict(params)
        handler = get_handlers(request.session_id).session_fs
        if handler is None: raise RuntimeError(f"No session_fs handler registered for session: {request.session_id}")
        result = await handler.read_file(request)
        return result.to_dict()
    client.set_request_handler("sessionFs.readFile", handle_session_fs_read_file)
    async def handle_session_fs_write_file(params: dict) -> dict | None:
        request = SessionFSWriteFileRequest.from_dict(params)
        handler = get_handlers(request.session_id).session_fs
        if handler is None: raise RuntimeError(f"No session_fs handler registered for session: {request.session_id}")
        result = await handler.write_file(request)
        return result.to_dict() if result is not None else None
    client.set_request_handler("sessionFs.writeFile", handle_session_fs_write_file)
    async def handle_session_fs_append_file(params: dict) -> dict | None:
        request = SessionFSAppendFileRequest.from_dict(params)
        handler = get_handlers(request.session_id).session_fs
        if handler is None: raise RuntimeError(f"No session_fs handler registered for session: {request.session_id}")
        result = await handler.append_file(request)
        return result.to_dict() if result is not None else None
    client.set_request_handler("sessionFs.appendFile", handle_session_fs_append_file)
    async def handle_session_fs_exists(params: dict) -> dict | None:
        request = SessionFSExistsRequest.from_dict(params)
        handler = get_handlers(request.session_id).session_fs
        if handler is None: raise RuntimeError(f"No session_fs handler registered for session: {request.session_id}")
        result = await handler.exists(request)
        return result.to_dict()
    client.set_request_handler("sessionFs.exists", handle_session_fs_exists)
    async def handle_session_fs_stat(params: dict) -> dict | None:
        request = SessionFSStatRequest.from_dict(params)
        handler = get_handlers(request.session_id).session_fs
        if handler is None: raise RuntimeError(f"No session_fs handler registered for session: {request.session_id}")
        result = await handler.stat(request)
        return result.to_dict()
    client.set_request_handler("sessionFs.stat", handle_session_fs_stat)
    async def handle_session_fs_mkdir(params: dict) -> dict | None:
        request = SessionFSMkdirRequest.from_dict(params)
        handler = get_handlers(request.session_id).session_fs
        if handler is None: raise RuntimeError(f"No session_fs handler registered for session: {request.session_id}")
        result = await handler.mkdir(request)
        return result.to_dict() if result is not None else None
    client.set_request_handler("sessionFs.mkdir", handle_session_fs_mkdir)
    async def handle_session_fs_readdir(params: dict) -> dict | None:
        request = SessionFSReaddirRequest.from_dict(params)
        handler = get_handlers(request.session_id).session_fs
        if handler is None: raise RuntimeError(f"No session_fs handler registered for session: {request.session_id}")