hevolus-xrcopilotlab 2.2.0

Python SDK for XRCopilotLab — multi-agent orchestration, agents, topics, profiles.

hevolus-xrcopilotlab

Async Python SDK for the XRCopilotLab Public API — multi-agent orchestrations, agents, topics, profiles, and common services.

Version

Current Version: 2.2.0

Features

• Topic Management: Create, update, delete topics and manage their files (upload, move, download, SAS URLs).
• Profile Management: Handle profiles, trigger Knowledge Graph sync/purge, check indexing status.
• Agent Management: CRUD agents, chat (sync and async fire-and-forget), conversation history, temp file uploads.
• Orchestrators (v2.2.0): List, get, and validate orchestrator workflows; dispatch executions (fire-and-forget); pause and resume on user interaction; stream live progress over Server-Sent Events; poll status; cancel; manage orchestrator-scoped files.
• Common Services: Access shared utilities — voices, avatars (2D and 3D), speech synthesis & recognition, news.

Installation

pip install hevolus-xrcopilotlab

Requires Python 3.10 or later.

Usage

Initialization

XRCopilotLabClient is an async context manager built on httpx. All network I/O is non-blocking and must be awaited inside an async function.

from xrcopilotlab import XRCopilotLabClient

# Default API version (2025-05-01)
async with XRCopilotLabClient(api_key="your-api-key") as client:
    ...

# Specific API version
async with XRCopilotLabClient(api_key="your-api-key", api_version="2025-06-01") as client:
    ...

# Staging environment
async with XRCopilotLabClient(api_key="your-api-key", api_version="staging") as client:
    ...

# Override base URL (e.g. local Azure Functions runtime)
async with XRCopilotLabClient(
    api_key="your-api-key",
    base_url="http://localhost:7071/api/",
) as client:
    ...

Accessing Services

The client exposes five service namespaces:

• client.topics — TopicService: topic CRUD and file management
• client.profiles — ProfileService: profile CRUD, Knowledge Graph, indexing status
• client.agents — AgentService: agent CRUD, chat (sync/async), conversation history, temp files
• client.orchestrators — OrchestratorService: orchestration lifecycle, SSE streaming, file management
• client.common — CommonService: voices, speech synthesis & recognition, avatars, news

---

Examples

List Topics

import asyncio
from xrcopilotlab import XRCopilotLabClient


async def main() -> None:
    async with XRCopilotLabClient(api_key="your-api-key") as client:
        topics = await client.topics.get_topics(company_id="company-id")
        for t in topics:
            print(t.topic_id, t.name)


asyncio.run(main())

Upload a File

import asyncio
from xrcopilotlab import XRCopilotLabClient


async def main() -> None:
    async with XRCopilotLabClient(api_key="your-api-key") as client:
        with open("document.pdf", "rb") as f:
            await client.topics.upload_file(
                file=f,
                company_id="company-id",
                topic_id="topic-id",
                file_name="document.pdf",
                scope=0,  # 0 = Profile, 1 = Agent
            )


asyncio.run(main())

Agent Chat (Sync) with Conversation Continuation

from xrcopilotlab.models import UserPrompt


async def chat(client):
    prompt = UserPrompt(
        company_id="company-id",
        topic_id="topic-id",
        agent_id="agent-id",
        endpoint_id="endpoint-id",
        message="Hello, how can you help me?",
    )
    response = await client.agents.chat(prompt)

    follow_up = UserPrompt(
        company_id="company-id",
        topic_id="topic-id",
        agent_id="agent-id",
        endpoint_id="endpoint-id",
        message="Tell me more.",
        conversation_id=response.conversation_id,
    )
    await client.agents.chat(follow_up)

Async Chat (Fire-and-Forget) with Polling

start_chat submits the prompt and returns immediately (HTTP 202). Poll get_conversation_history until an assistant message appears.

import asyncio
from xrcopilotlab.models import UserPrompt


async def async_chat(client):
    prompt = UserPrompt(
        company_id="company-id",
        topic_id="topic-id",
        agent_id="agent-id",
        endpoint_id="endpoint-id",
        message="Analyze this data in detail",
    )
    accepted = await client.agents.start_chat(prompt)

    history = None
    while not history or not any(m.role == "assistant" for m in history.messages):
        await asyncio.sleep(2)
        history = await client.agents.get_conversation_history(
            company_id=accepted.company_id,
            topic_id=accepted.topic_id,
            agent_id=accepted.agent_id,
            conversation_id=accepted.conversation_id,
            messages=0,
        )

    print("Agent replied:", history.messages[-1].content)

Orchestrator: High-Level (execute_and_wait)

execute_and_wait starts the orchestration, subscribes to its SSE event stream, calls interaction_handler whenever the workflow pauses for user input, and returns the final OrchestrationResult:

import asyncio
from xrcopilotlab import XRCopilotLabClient
from xrcopilotlab.models import (
    OrchestrationRequest,
    UserInteractionRequest,
    UserInteractionResponse,
)


async def main() -> None:
    async with XRCopilotLabClient(api_key="your-api-key") as client:

        async def on_interaction(req: UserInteractionRequest) -> UserInteractionResponse:
            return UserInteractionResponse(
                request_id=req.interaction_id,
                response_value="yes",
                answer_language="en",
            )

        result = await client.orchestrators.execute_and_wait(
            company_id="company-id",
            orchestrator_id="orchestrator-id",
            request=OrchestrationRequest(message="Analyze Q3 sales"),
            interaction_handler=on_interaction,
            progress_handler=lambda evt: print(f"[{evt.seq}] {evt.event_type}"),
        )
        print("Final:", result.final_output)


asyncio.run(main())

Orchestrator: Low-Level (stream_execution + manual resume_execution)

For full lifecycle control, drive start_execution → stream_execution → resume_execution yourself:

from xrcopilotlab import OrchestrationEventTypes
from xrcopilotlab.models import OrchestrationRequest, UserInteractionResponse


async def low_level(client, company_id, orchestrator_id):
    handle = await client.orchestrators.start_execution(
        company_id=company_id,
        orchestrator_id=orchestrator_id,
        request=OrchestrationRequest(message="Analyze Q3 sales"),
    )

    async for evt in client.orchestrators.stream_execution(
        company_id=company_id, execution_id=handle.execution_id,
    ):
        if evt.event_type == OrchestrationEventTypes.USER_INTERACTION_REQUIRED.value:
            req = client.orchestrators.extract_interaction_request(evt)
            await client.orchestrators.resume_execution(
                interaction_id=req.interaction_id,
                orchestration_state=req.orchestration_state,
                user_response=UserInteractionResponse(
                    request_id=req.interaction_id,
                    response_value="yes",
                    answer_language="en",
                ),
            )
            continue
        if evt.event_type in {
            OrchestrationEventTypes.COMPLETED.value,
            OrchestrationEventTypes.FAILED.value,
            OrchestrationEventTypes.CANCELLED.value,
        }:
            break

Orchestrator: File Management

Each orchestration run can read files uploaded to its orchestrator-scoped storage. Set OrchestrationRequest.temp_files = True to pass them to the workflow.

import asyncio
from xrcopilotlab import XRCopilotLabClient


async def main() -> None:
    async with XRCopilotLabClient(api_key="your-api-key") as client:
        # Upload (max 100 MB per file in v1)
        with open("input.csv", "rb") as f:
            await client.orchestrators.upload_file(
                file=f,
                company_id="company-id",
                orchestrator_id="orchestrator-id",
                file_name="input.csv",
                scope=1,
            )

        # Read-only SAS URL (valid for ~1 hour)
        url = await client.orchestrators.get_file_read_url(
            company_id="company-id",
            orchestrator_id="orchestrator-id",
            file_name="input.csv",
        )
        print("Download URL:", url)

        # Delete when no longer needed
        await client.orchestrators.delete_file(
            company_id="company-id",
            orchestrator_id="orchestrator-id",
            file_name="input.csv",
        )


asyncio.run(main())

Knowledge Graph Sync (Profile-Level)

import asyncio
from xrcopilotlab import XRCopilotLabClient


async def main() -> None:
    async with XRCopilotLabClient(api_key="your-api-key") as client:
        # Trigger ingestion for all profile files
        await client.profiles.sync_knowledge_graph(
            company_id="company-id",
            topic_id="topic-id",
            profile_id="profile-id",
        )

        # Check indexing status
        status = await client.profiles.get_indexing_status(
            company_id="company-id",
            topic_id="topic-id",
            profile_id="profile-id",
        )
        print("Indexing status:", status)

        # Purge all Knowledge Graph data for a profile
        await client.profiles.purge_knowledge_graph(
            company_id="company-id",
            topic_id="topic-id",
            profile_id="profile-id",
        )


asyncio.run(main())

Common Services

import asyncio
from xrcopilotlab import XRCopilotLabClient
from xrcopilotlab.models import SpeechRequest, SpeechRecognitionRequest


async def main() -> None:
    async with XRCopilotLabClient(api_key="your-api-key") as client:
        # Available voices
        voices = await client.common.get_voices()
        print("Voices:", [v.voice_id for v in voices])

        # Text to speech
        speech_req = SpeechRequest(text="Hello world", language="en", voice_id="voice-id")
        audio = await client.common.speech(speech_req)

        # Speech to text
        recognition_req = SpeechRecognitionRequest(
            audio_data="<base64-encoded-audio>",
            language="en-US",
            format="audio/webm",
        )
        recognized = await client.common.speech_recognition(recognition_req)
        print("Transcription:", recognized.text, "| Confidence:", recognized.confidence)

        # Avatars
        avatars = await client.common.get_avatars(company_id="company-id")
        avatars_3d = await client.common.get_3d_avatars(company_id="company-id")

        # News
        news = await client.common.get_news()


asyncio.run(main())

---

API Reference

TopicService

Method	Description
get_topics	Get all topics for a company
get_topic_list	Get topic list (lightweight, no nested profiles/agents)
get_topic	Get a specific topic by ID
save_topic	Create or update a topic
delete_topic	Delete a topic
check_topic_exists	Check if a topic exists by ID or name
upload_file	Upload a file to a topic
delete_file	Delete a file from a topic
get_file_url	Get the URL of a file
get_file_read_url	Get a read-only URL for a file
get_file_sas_url	Get a SAS URL for direct file access
get_file_stream	Download a file as a binary stream
move_file	Move a file within a topic
download_folder_as_base64	Download multiple files packed as a Base64-encoded ZIP

ProfileService

Method	Description
get_profiles	Get all profiles for a topic
get_profile_list	Get profile list (lightweight)
get_profile	Get a specific profile by ID
save_profile	Create or update a profile
delete_profile	Delete a profile
activate_profile	Activate or deactivate a profile
delete_profile_file	Delete a file from a profile
get_indexing_status	Get indexing status for a profile
sync_knowledge_graph	Trigger Knowledge Graph ingestion for profile files
purge_knowledge_graph	Purge all Knowledge Graph data for a profile

AgentService

Method	Description
get_agents	Get all agents for a topic
get_agent_list	Get agent list (lightweight)
get_agent	Get a specific agent by ID
save_agent	Create or update an agent
delete_agent	Delete an agent
activate_agent	Activate or deactivate an agent
get_welcome_message	Get the welcome message text for an agent
get_full_welcome_message	Get the full welcome message response object
chat	Send a chat message and wait for the response
start_chat	Submit a chat prompt asynchronously (returns 202); poll with get_conversation_history
get_conversation_history	Retrieve conversation history by conversation ID
upload_temp_file	Upload a temporary file for use in a chat turn
get_temp_files_status	Get the processing status of uploaded temp files
delete_temp_file	Delete a temporary file

OrchestratorService

Method	Description
get_orchestrators	List orchestrators for a company
get_orchestrator	Get a specific orchestrator (full graph: steps, connections, endpoints)
validate_orchestrator	Validate an orchestrator definition before save/run
start_execution	Dispatch an orchestrator run (returns 202 Accepted + ExecutionHandle)
resume_execution	Resume a paused execution with the user's response
cancel_execution	Request cancellation of a running execution (idempotent)
get_execution_status	Snapshot status (status, last_seq, result when terminal)
stream_execution	Async generator over SSE events; auto-reconnects on soft-close
subscribe	Callback-style SSE subscription; returns a handle with a close() method
execute_and_wait	High-level wrapper: starts, streams, prompts on interaction, returns final OrchestrationResult
extract_interaction_request	Map a userInteractionRequired event onto a typed UserInteractionRequest
upload_file	Upload a file to orchestrator-scoped storage (max 100 MB)
get_file_read_url	Read-only SAS URL for an orchestrator file (valid ~1 hour)
delete_file	Delete a file from orchestrator-scoped storage

Execution Model

start_execution and resume_execution return immediately with an ExecutionHandle (HTTP 202). The orchestration runs asynchronously in the background. To observe progress, either subscribe to the SSE stream (stream_execution) or poll get_execution_status until it reaches a terminal state. execute_and_wait wraps the full execute → stream → prompt → result loop in a single awaitable call.

OrchestrationEvent.event_type Values

Value	When emitted
started	Execution accepted by the worker
stepStarted / stepCompleted	Step lifecycle (evt.step_id populated)
agentUpdate	Per-agent progress within a step (evt.step_id and evt.agent_id populated)
progress	Free-form markers; agent step completions emit output, agent_name, display_in_chat
userInteractionRequired	Execution paused; pass interaction_id + orchestration_state to resume_execution
paused / resumed	Pause / resume lifecycle markers
completed / failed / cancelled	Terminal — the SSE stream closes after these
heartbeat	Keep-alive only; ignore in business logic

CommonService

Method	Description
get_voices	Get available text-to-speech voices
speech	Convert text to speech audio
speech_recognition	Convert speech audio (Base64) to text
get_avatars	Get available human avatars for a company
get_3d_avatars	Get available 3D avatars for a company
get_news	Get news items

---

Configuration

Parameter	Description
api_key	Required. Passed as the Ocp-Apim-Subscription-Key HTTP header on every request.
api_version	API version string (e.g. "2025-05-01", "staging"). Appended as the api-version query parameter. Defaults to "2025-05-01".
base_url	Override the API base URL. Useful for local development against the Azure Functions runtime at http://localhost:7071/api/.

Local Development

To point the SDK at a locally running Azure Functions host:

async with XRCopilotLabClient(
    api_key="your-api-key",
    base_url="http://localhost:7071/api/",
) as client:
    ...

---

Dependencies

Package	Minimum Version	Purpose
httpx	>=0.27	Async HTTP client
httpx-sse	>=0.4	Server-Sent Events support
pydantic	>=2.6	Model validation and serialization
Python	3.10+	Runtime (uses match, `

---

Notes / Limitations

• Orchestrator file size limit: In v1, upload_file on OrchestratorService raises XRCopilotApiError(413) for files 100 MB or larger. Streaming uploads for very large files are deferred to a future release.
• SSE auto-reconnect: stream_execution transparently reconnects on soft-close (empty data: frames or network hiccups) using the last received seq as a resume cursor. Hard disconnects will surface as httpx.ReadError and should be caught by the caller.
• Conversation IDs: If conversation_id is not provided in UserPrompt, a UUID is auto-generated by the SDK. To continue an existing conversation, pass the conversation_id returned in the previous response.
• Thread safety: Each XRCopilotLabClient instance holds a single httpx.AsyncClient. Do not share a client instance across multiple event loops.

---

Changelog

Version 2.2.0

Initial release of the Python SDK.

• New Package: hevolus-xrcopilotlab — async Python SDK (XRCopilotLabClient) mirroring the .NET and JavaScript public SDKs.
• Services: TopicService (14 methods), ProfileService (10 methods), AgentService (14 methods), OrchestratorService (14 methods), CommonService (6 methods) — 58 methods total.
• Orchestrators: Full orchestration lifecycle including start_execution (fire-and-forget, 202 Accepted), stream_execution (SSE async generator with auto-reconnect), execute_and_wait (high-level one-call wrapper), resume_execution (user interaction), cancel_execution (idempotent), get_execution_status (snapshot polling), and orchestrator-scoped file management.
• Models: All request/response types — OrchestrationRequest, OrchestrationResult, ExecutionHandle, ExecutionStatus, OrchestrationEvent, UserInteractionRequest, UserInteractionResponse, UserPrompt, SpeechRequest, SpeechRecognitionRequest, and more.
• Enums: OrchestrationEventTypes covering the full set of SSE event type values.

---

For the full SDK reference (in .NET, JavaScript, and Python), see docs/XRCopilotLab-SDK-Documentation.md.