Python SDK — CosmosClient Reference
Main Python SDK client, authentication flows, sub-clients, registry, LLM, permissions, events, and lifecycle
Updated 2026-11-03
Verified 2026-11-03
CosmosClient is the main Python SDK entry point. It wires authentication, HTTP clients, ADK, the Configurator registry, Storage, LLM management, local tools, permissions, events, observability, and optional RAGflow search into one async client.
Configuration
from cosmos_sdk import AuthTokens, CosmosClient, CosmosClientConfig
config = CosmosClientConfig(
org_id="your-org-uuid",
services={
"user": "https://<user-service-host>",
"configurator": "https://<configurator-host>",
"storage": "https://<storage-host>",
"adk": "https://<adk-host>",
},
auth_tokens=AuthTokens(
access_token="<jwt>",
refresh_token="<refresh-token>",
expires_at=1999999999.0,
),
)
client = CosmosClient(config)
Each Cosmos service is reached at its own URL. For local development, use the service ports directly (User http://localhost:8101, Configurator http://localhost:8100, MCP tools http://localhost:8102, ADK http://localhost:8103, Storage http://localhost:8107, RAGflow http://localhost:8200). Add "ragflow" to services to enable client.ragflow.
gateway_url is an alternative only when a deployment exposes every service behind one host. Gateway mode appends service prefixes such as /user and /adk, so do not set it to one service host such as https://<adk-host>; that produces invalid URLs. See Base URLs and Required headers.
Constructors
| Constructor | Use When |
|---|---|
await CosmosClient.from_credentials(...) | Script or CLI logs in with email/password |
CosmosClient.from_token(...) | Web framework already has an access token |
await CosmosClient.from_oauth_code(...) | OAuth2 PKCE redirect callback |
CosmosClient(CosmosClientConfig(...)) | Full control over config and tokens |
Sub-Clients
| Property | Purpose |
|---|---|
client.auth | Token lifecycle, refresh, JWT decode helpers |
client.authz | User Service authorization decision helpers |
client.registry | Configurator registry for agents, tools, skills, templates, apps, and collections |
client.adk | REST converse, ADK V2 WebSocket, sessions, async requests, skill generation |
client.storage | Cosmos Storage / DMS files and folders |
client.llm | Client-side provider routing, model discovery, BYOT metadata |
client.tool_manager | Local tools, MCP server connections, hooks, batch execution |
client.permissions | Allow/deny/prompt policies for tool access |
client.observability | Event batching and adapter-based analytics export |
client.events | SDK event emitter |
client.ragflow | Optional RAGflow search client when ragflow URL is configured |
Service-to-Service Authentication
For backend services, background workers, and automated pipelines that don't have a user session, the SDK supports OAuth2 client_credentials authentication.
ServiceAuthManager
Use ServiceAuthManager directly when you need fine-grained control over the token lifecycle:
from cosmos_sdk import ServiceAuthManager
auth = ServiceAuthManager(
"https://<user-service-host>",
"your-org-uuid",
client_id="your-client-id",
client_secret="<client-secret>",
scope="read write", # optional, defaults to "read"
audience="cosmos-storage", # optional, defaults to client_id
)
token = await auth.get_access_token() # returns cached or freshly acquired JWT
get_access_token() acquires a token on first use, caches it, and re-grants it before expiry. Subsequent calls return the cached value until it is near expiration. The Python SDK always sends an audience; when none is given it uses the client ID.
Factory Method for Long-Lived Clients
Use CosmosClient.from_client_credentials() when you want a fully wired client that auto-refreshes its own service tokens:
from cosmos_sdk import CosmosClient
client = await CosmosClient.from_client_credentials(
"your-client-id",
"<client-secret>",
org_id="your-org-uuid",
services={
"user": "https://<user-service-host>", # token endpoint: <user>/oauth/token/
"configurator": "https://<configurator-host>",
"adk": "https://<adk-host>",
},
scope="read write",
audience="cosmos-adk",
)
# Use exactly like a user-authenticated client
agents = await client.search_agents({"query": "support"})
await client.close()
The resulting token has actor_type: "service" and identifies the caller as the service, not an individual user.
When to Use Each
| Approach | Best for |
|---|---|
ServiceAuthManager | One-off scripts, custom HTTP calls, or when you manage the client yourself |
CosmosClient.from_client_credentials() | Long-lived services, background workers, cron jobs that make many SDK calls |
Registry Facade
The facade normalizes paginated responses into Page(items, total, page, page_size) and caches read operations in the registry manager for 5 minutes.
agents = await client.search_agents({"query": "sales", "page_size": 10})
tools = await client.search_tools({"query": "weather"}, tool_type="api")
skills = await client.search_skills({"tags": ["finance"]})
templates = await client.search_templates()
collections = await client.search_collections()
Data-source helpers are facade methods as well:
sources = await client.search_data_sources({"query": "warehouse", "engine": "sqlite"})
source = await client.create_data_source({
"name": "Assessment SQLite",
"engine": "sqlite",
"connection_profile": {"path": "/dms/projects/acme/assessment.db"},
})
await client.sync_data_source_tool(source["id"])
test = await client.test_data_source(source["id"], query="SELECT 1")
print(sources.total, test)
Skills and collections support write operations through the SDK:
skill = await client.create_skill({
"name": "refund-policy",
"description": "Handle refund policy questions",
"configuration": {
"instructions": "Use the refund policy collection before answering.",
},
})
await client.deploy_skill(skill["id"])
collection = await client.create_collection({
"name": "Client Policies",
"description": "Uploaded policy documents",
})
await client.reindex_collection(collection["id"])
Agents, tools, templates, and apps are exposed as read/search helpers for application integration. Use Configurator APIs or the Cosmos frontend for authoring those resources.
LLM Manager
client.llm discovers model and BYOT configuration from Configurator and routes complete() calls through locally registered Python providers.
models = await client.fetch_available_models()
org_models = await client.fetch_org_models()
byot_schema = await client.fetch_byot_schema()
byot_config = await client.fetch_byot_config()
Provider routing is useful when your app owns the actual provider call but wants Cosmos policy and model catalog integration.
User And Role Helpers
roles = await client.fetch_org_roles(expand="permissions")
role = await client.fetch_role_detail("role-id")
assignments = await client.fetch_org_user_roles()
Authorization Decisions
allowed = await client.authz.can({
"actor": {"type": "user", "id": "user-uuid"},
"action": "bridge.connect",
"resource": {"type": "organization", "id": "org-uuid"},
"context": {"channel": "jumpstart"},
})
Tools And MCP
Register local tools for ADK V2 callbacks:
from cosmos_sdk import ToolDefinition, ToolResult
async def lookup_customer(args: dict) -> ToolResult:
return ToolResult(ok=True, data={"customer_id": args["email"]})
client.tool_manager.register_tool(
"lookup_customer",
handler=lookup_customer,
capabilities=["crm.customer.lookup"],
definition=ToolDefinition(
name="lookup_customer",
description="Look up a customer by email",
input_schema={
"type": "object",
"properties": {"email": {"type": "string"}},
"required": ["email"],
},
),
)
Connect to MCP servers when the app needs external tool catalogs:
from cosmos_sdk import MCPServerConfig
await client.tool_manager.connect_mcp(
MCPServerConfig(server_id="cosmos", url="https://<mcp-tools-host>/mcp")
)
Events And Observability
send_message() emits SDK lifecycle events such as conversation.started, tool.start, tool.result, and agent.iteration.
client.events.on("tool.start", lambda event: print(event.payload))
client.events.on_prefix("tool.", lambda event: audit(event))
Use client.observability when an application needs to batch and flush SDK-side events to its own analytics adapter.
Document Tools
create_document_tools() (exported from cosmos_sdk) builds read, create, append, and edit tools over a document store. Write tools reject payloads that contain projected artifact-replay content (a placeholder ADK substitutes for large earlier tool output) with TOOL_INVALID_INPUT, instead of persisting an incomplete file. The TypeScript SDK applies the same guard.
Lifecycle
await client.close()
shutdown is an alias for close. Use async with CosmosClient(config) as client: when possible.