Sign in

Python SDK — Quick Start

Install, configure, and make your first Cosmos agent call from Python

sdkpythonquickstartgetting-started

Updated 2026-11-03

Verified 2026-11-03

cosmos-sdk-python is the async Python client for Cosmos. Use it when a backend service, notebook, CLI, or worker needs to authenticate with Cosmos, run ADK agents, use client-side tools, read and write DMS files, query the registry, or discover configured models.

Prerequisites

  • Python 3.10 or later
  • A Cosmos organization ID
  • Either a Cosmos access token or a user login that can access the organization
  • Direct Cosmos service URLs, or one real gateway URL that routes to all service prefixes

Installation

bash
pip install git+https://github.com/ey-org/cosmos-sdk-python.git@develop

For local SDK development, use uv run pytest inside cosmos-sdk-python/.

Configure URLs

Prefer direct service URLs when the deployment exposes service-specific hosts:

bash
export COSMOS_ORG_ID="your-org-uuid"
export COSMOS_USER_SERVICE_URL="https://user.example.com"
export COSMOS_CONFIGURATOR_SERVICE_URL="https://configurator.example.com"
export COSMOS_ADK_SERVICE_URL="https://adk.example.com"
export COSMOS_STORAGE_SERVICE_URL="https://storage.example.com"

Gateway mode is only for a real gateway that routes /user, /configurator, /adk, and other service prefixes:

bash
export COSMOS_GATEWAY_URL="https://cosmos.example.com/cosmos"

Do not set COSMOS_GATEWAY_URL to one service host such as https://adk.example.com. That makes the SDK derive invalid URLs like https://adk.example.com/user.

For local development, point each variable at the service port directly, for example COSMOS_USER_SERVICE_URL=http://localhost:8101, COSMOS_CONFIGURATOR_SERVICE_URL=http://localhost:8100, COSMOS_ADK_SERVICE_URL=http://localhost:8103, and COSMOS_STORAGE_SERVICE_URL=http://localhost:8107.

services entries override gateway_url for that service. cosmos_base_url is retained only as a backward-compatible alias for gateway_url.

Service Authentication

For backend services and workers that authenticate with OAuth2 client credentials instead of a user token:

python
import os
from cosmos_sdk import CosmosClient, build_service_endpoints_from_env

client = await CosmosClient.from_client_credentials(
    "your-client-id",
    os.environ["COSMOS_CLIENT_SECRET"],
    org_id=os.environ["COSMOS_ORG_ID"],
    services=build_service_endpoints_from_env(),  # must include the User Service URL
)

The client acquires and auto-refreshes service tokens. See CosmosClient Reference — Service-to-Service Authentication for ServiceAuthManager and advanced options.

First Agent Call

Use from_token() when your app already owns authentication, such as a FastAPI or Django service receiving a Cosmos JWT.

python
import asyncio
import os

from cosmos_sdk import CosmosClient
from cosmos_sdk import build_service_endpoints_from_env


async def main() -> None:
    services = build_service_endpoints_from_env()
    gateway_url = os.getenv("COSMOS_GATEWAY_URL") or None

    client = CosmosClient.from_token(
        os.environ["COSMOS_ACCESS_TOKEN"],
        org_id=os.environ["COSMOS_ORG_ID"],
        services=services,
        gateway_url=gateway_url,
        verify_tls=True,
    )

    try:
        claims = client.auth.decode_token()
        conversation_user_id = claims.sub or claims.user_id

        result = await client.adk.converse(
            query="What can this agent help with?",
            user_id=conversation_user_id,
            agent_id=os.environ["COSMOS_AGENT_ID"],
            agent_version=os.getenv("COSMOS_AGENT_VERSION", "1.0"),
            channel="chat",
        )
        print(result["response"]["content"])
        print("Session:", result["session_id"])
    finally:
        await client.close()


asyncio.run(main())

Login From a Script

Use from_credentials() for CLI scripts and controlled automation. Do not hardcode credentials in source files.

python
import asyncio
import os

from cosmos_sdk import CosmosClient
from cosmos_sdk import build_service_endpoints_from_env


async def main() -> None:
    services = build_service_endpoints_from_env()
    gateway_url = os.getenv("COSMOS_GATEWAY_URL") or None

    client = await CosmosClient.from_credentials(
        os.environ["COSMOS_EMAIL"],
        os.environ["COSMOS_PASSWORD"],
        org_id=os.environ["COSMOS_ORG_ID"],
        services=services,
        gateway_url=gateway_url,
    )

    try:
        agents = await client.search_agents({"query": "support", "page_size": 5})
        for agent in agents.items:
            print(agent["name"], agent["id"])
    finally:
        await client.close()


asyncio.run(main())

Streaming With ADK V2

Use client.send_message() for WebSocket streaming, client-side tool callbacks, and multi-turn sessions.

python
from cosmos_sdk import ADKV2SendOptions, ADKV2TurnCallbacks


callbacks = ADKV2TurnCallbacks(
    on_chunk=lambda text: print(text, end="", flush=True),
    on_tool_start=lambda name: print(f"\nTool started: {name}"),
    on_error=lambda code, message, fatal: print(f"\n{code}: {message}"),
)

claims = client.auth.decode_token()
conversation_user_id = claims.sub or claims.user_id

result = await client.send_message(
    "Summarize the uploaded policy documents.",
    user_id=conversation_user_id,
    callbacks=callbacks,
    options=ADKV2SendOptions(
        agent_id="agent-uuid",
        agent_version="1.0",
        reasoning_event=True,
        session_state={"project_id": "project-uuid"},
    ),
)

print("\nSession:", result.session_id)
client.close_adk_session()

Clean Up

Always close the client in long-running processes and tests. This closes HTTP clients, WebSocket sessions, and observability flush tasks.

python
await client.close()

CosmosClient also supports async context management:

python
async with CosmosClient(config) as client:
    ...

Next Steps