dt

dt is a Python SDK for Dynatrace. It covers the platform APIs (DQL queries, documents, workflows, Settings, and more), the classic environment APIs, Managed clusters, and account management. Every operation is a typed method documented in your IDE, every response is a model, and the same code runs synchronously or with asyncio.

Install

Python 3.12 or later.

pip install dt>=2.0.0a1

To build a wheel or work on the SDK itself, see BUILD.md.

Connect

export DT_ENV_URL=https://abc12345.apps.dynatrace.com
export DT_PLATFORM_TOKEN="<platform token>"

from dt import PlatformClient

client = PlatformClient()

Each client reads its URL and credentials from its own environment variables:

A client takes the first of its URL variables that is set: DT_PLATFORM_URL before DT_ENV_URL for PlatformClient, and DT_CLASSIC_URL before DT_ENV_URL for ClassicClient. DT_ENVIRONMENT_URL is an alias of DT_ENV_URL, read after it.

A classic URL looks like https://abc12345.live.dynatrace.com. ClassicClient and ClusterClient take a token only, because their URLs don't accept OAuth; to call the classic environment APIs with platform credentials, use PlatformClient().classic (see Explore the API). OAuth, and bearer_token= for a token you already have, are for PlatformClient and AccountClient. With OAuth, a call requests the scopes its operation declares. DQL queries declare none, because their scopes depend on the data they read: list those in scopes= or DT_SCOPES, such as storage:logs:read storage:buckets:read.

Arguments override the environment:

client = PlatformClient("https://abc12345.apps.dynatrace.com", platform_token="<token>")

print(client) shows what a client connects to and which credential argument it uses, passed or from the environment, never the token itself: PlatformClient(url='https://abc12345.apps.dynatrace.com', platform_token='***').

A client keeps its connections open for reuse and is safe to share between threads, so create one and reuse it. client.close(), or the end of a with PlatformClient() as client: block, closes them; a closed client raises RuntimeError when it would send a request. A client you don't close keeps its connections until Python garbage-collects it or the program ends.

Run a DQL query

result = client.query("fetch logs | summarize count(), latest = max(timestamp), by: {loglevel}")
for record in result.records:
    print(record["loglevel"], record["count()"], repr(record["latest"]))
# ERROR 1234 datetime.datetime(2026, 10, 15, 9, 41, 7, 512000, tzinfo=datetime.timezone.utc)
# INFO 98765 datetime.datetime(2026, 10, 15, 9, 41, 59, 4000, tzinfo=datetime.timezone.utc)

query() returns once the query has finished (it waits up to wait_timeout, 300 seconds by default), and each record is a dict of Python values: a long is an int, a timestamp a datetime, a duration a timedelta, and a timeframe a Timeframe with start and end. Query options are keyword arguments, such as default_timeframe_start, default_timeframe_end, max_result_records, and filter_segments.

Send data to Grail

from dt import ClassicClient

records = [{"content": "Card declined", "loglevel": "WARN", "log.source": "checkout"}]
ClassicClient().environment.v2.logs.create(body=records)  # environment URL, access token
client.classic.environment.v2.logs.create(body=records)  # platform URL, platform token

Dynatrace has no separate Grail ingest API: the environment API's ingest endpoints feed OpenPipeline, which stores the data in Grail. ClassicClient sends to them on the environment URL with an access token; PlatformClient sends to the same endpoints on the platform URL, under classic, with a platform token or OAuth. Every ingest is a create() method: logs and business events take one record or a list of them (logs.create, bizevents.create), and metrics take the line protocol (metrics.create). Two kinds only ClassicClient reaches: events through the OpenPipeline ingest API (openpipeline.v1.events.create, and SDLC and security events beside it) and spans over OTLP (environment.v2.otlp.v1.traces.create). See logs, events, and OpenTelemetry.

List and get

for doc in client.document.v1.documents.list(filter="type == 'dashboard'"):
    print(doc.id, doc.name)
# 6f3c2d9e-1b4a-4e8f-9c0d-2a7b5e1f8c34 Checkout overview
# a91d7c2e-58f0-4b3e-8d6a-0c4e9f2b7a15 Kubernetes cluster health

doc = client.document.v1.documents.metadata.get("6f3c2d9e-1b4a-4e8f-9c0d-2a7b5e1f8c34")
print(doc.name, doc.owner)
# Checkout overview 9b1f0c3e-7a2d-4e65-b8c4-1d0a2f6e9b37

list() yields every item and fetches the next page when the loop reaches it, so leaving the loop early fetches nothing more. What list() returns is the first page: .items, .has_more, and .next_page() work page by page. Path parameters, such as ids, are positional, and a parameter named like a Python keyword ends in an underscore (from_).

A read without an id is a list() too, also where the API returns a single object, such as a configuration; then list() returns that object, not a page. ClassicClient().environment.v2.active_gates.auto_update.list() returns the global auto-update configuration of ActiveGates, and ClusterClient().v1.smtp.list() the SMTP configuration.

Create and update

doc = client.document.v1.documents.create(
    name="On-call notes", type="note", content=b"Restart the checkout service first."
)
print(doc.id, doc.version)
# 0b7e5c1a-9d24-4f6e-b3a8-5c2f1e7d9a40 1

updated = client.document.v1.documents.update(
    doc.id, optimistic_locking_version=doc.version, labels=["on-call"]
)
print(updated.document_metadata.version)
# 2

workflow = client.automation.v1.workflows.create(title="Nightly cleanup", type="SIMPLE")
print(workflow.id, workflow.type)
# 3c8e1f42-6b7d-4a90-9e25-d1f0a7b4c863 SIMPLE

The fields of a request body are keyword arguments, enumeration arguments take plain strings such as "SIMPLE", and your IDE completes both. A create call returns the new object with the id Dynatrace gave it. Nested objects are models (see Models below), and a body that isn't an object, such as a list, is passed as body=.

Where an API does the same thing to all items (or to a global setting) and to one item, one method does both, with the item's id as an optional positional argument: active_gates.auto_update.update() changes the global auto-update configuration of ActiveGates, and update(ag_id) that of one ActiveGate. Without its id, such a delete() may delete every item, as client.state.v1.app_states.delete() deletes all states of an app; its docstring says so.

Settings

from dt.settings.builtin.ownership import Team

team = Team(
    name="Checkout",
    identifier="checkout",
    responsibilities=Team.Responsibilities(
        development=True,
        security=False,
        operations=True,
        infrastructure=False,
        line_of_business=False,
    ),
    contact_details=[Team.ContactDetails(integration_type="SLACK", slack_channel="#checkout")],
)
client.settings.v1.objects.create(team, scope="environment")
print(team.object_id)  # create() bound the model to the new object
# vu9U3hXa3q0AAAABABdidWlsdGluOm93bmVyc2hpcC50ZWFtcwAGdGVuYW50AAZ0ZW5hbnQAJDdlMjRm...

for other in client.settings.v1.objects.list(Team, scope="environment"):
    print(other.identifier, other.name)
# checkout Checkout
# payments Payments

team.description = "Owns the checkout and cart services"
client.settings.v1.objects.update(team)  # the model remembers its object id and version
client.settings.v1.objects.delete(team)

A Settings model is named after its schema ID: builtin:ownership.teams is dt.settings.builtin.ownership.Team, builtin:management-zones is dt.settings.builtin.ManagementZone, and builtin:tags.auto-tagging is dt.settings.builtin.tags.AutoTagging. Its nested types and enumerations are attributes of the model, such as Team.Responsibilities and Team.IntegrationType; enumeration fields also take plain strings such as "SLACK". These packages export the latest schema version; modules such as dt.settings.builtin.tags.v1_0_18 pin one. Without a model, get(object_id) and list(schema_id=..., scope=...) return untyped objects.

ClassicClient reads and writes the same models, with an access token that has the scopes settings.read and settings.write:

from dt import ClassicClient
from dt.settings.builtin import ManagementZone

objects = ClassicClient().environment.v2.settings.objects
zone = ManagementZone(name="Checkout", description="Hosts of the checkout service")
objects.create(zone, scope="environment")
for other in objects.list(ManagementZone, scope="environment"):
    print(other.name, other.description)
# Checkout Hosts of the checkout service
# Production None
objects.delete(zone)

There, a model from get() or list() carries the object's update token instead of a version: update() and delete() send it and fail if the object changed since. A write leaves the model without a token, so read the object again to keep that check for the next write.

Async

import asyncio

from dt import AsyncPlatformClient

async_client = AsyncPlatformClient()  # no event loop needed yet


async def main() -> None:
    result = await async_client.query("fetch logs | summarize count()")
    print(result.records)
    # [{'count()': 1043551}]
    async for doc in async_client.document.v1.documents.list():
        print(doc.name)
    # Checkout overview
    # Kubernetes cluster health
    await async_client.close()


asyncio.run(main())

Every client has an Async twin with the same methods: await the calls, and iterate list() with async for. As with the other clients, create one and reuse it, at module level or when your app starts. It runs in whichever event loop calls it, one loop at a time, so several asyncio.run() calls or a test per event loop can share it. await client.close(), or async with, closes its connections; close it in the event loop that used it last, such as at the end of main().

Errors

from dt.errors import ApiError, NotFoundError

try:
    doc = client.document.v1.documents.metadata.get("6f3c2d9e-1b4a-4e8f-9c0d-2a7b5e1f8c34")
except NotFoundError:
    doc = None
except ApiError as error:
    print(error.status_code, error.message, error.missing_scopes)
    # 403 The authorization token does not provide the necessary permissions. ['document:documents:read']

Everything the SDK raises derives from dt.errors.DynatraceError:

DynatraceError
├── ApiError                  an error response: status_code, message, constraint_violations
│   ├── BadRequestError       400
│   ├── AuthenticationError   401
│   ├── ForbiddenError        403
│   ├── NotFoundError         404
│   ├── ConflictError         409
│   ├── TooManyRequestsError  429 that outlasted the retries
│   └── ServerError           5xx
├── TransportError            no response, such as a connection failure or timeout
├── InvalidResponseError      a successful response that does not match its model
├── QueryError                a DQL query that failed or was canceled while running
│   └── QueryTimeoutError     a DQL query still running after wait_timeout
└── ConfigurationError        a missing or conflicting URL or credential

Invalid values raise pydantic's ValidationError before anything is sent. A 403 names what is missing where Dynatrace says so: error.missing_scopes lists the OAuth scopes the token lacks, and error.missing_permissions the IAM permissions its user lacks. With OAuth, a client that may not request a scope fails with an AuthenticationError that names the scopes it asked for.

Models

Responses are pydantic models with snake_case fields. Enum values compare equal to their strings (execution.state == "SUCCESS"), and fields or values that this SDK doesn't know yet are kept rather than rejected. Model fields take enumeration values as plain strings too (ScheduleRequest(timezone="Europe/Vienna")), and a field that allows a single value is filled in (CronTrigger(cron="0 8 * * 1-5")). A call that Dynatrace may answer without a body returns None then: a create-or-update PUT returns the new object when it creates one and None when it updates. Import a model to annotate your code or to build a nested request value; the import path follows the attribute path, down to the API version:

from dt.platform.document.v1 import DocumentMetaData  # PlatformClient().document.v1
from dt.platform.automation.v1 import Task  # PlatformClient().automation.v1
from dt.classic.environment.v2 import Entity  # ClassicClient().environment.v2
from dt.cluster.v2 import ActiveGate  # ClusterClient().v2
from dt.account import User  # AccountClient(), for all of its APIs
from dt.settings.builtin.ownership import Team  # Settings schema builtin:ownership.teams

Caching, retries, and timeouts

from datetime import timedelta

from dt import DefaultResponseCache, PlatformClient

client = PlatformClient(
    response_cache=DefaultResponseCache("~/.cache/dt.sqlite", max_age=timedelta(minutes=10)),
    timeout=30,
    max_retries=5,
)

The cache answers repeated GET requests and DQL queries from a SQLite file and never stores writes. timeout is the read timeout of each request in seconds (every method also takes timeout=). max_retries (3 by default) limits the retries of throttled requests and of idempotent requests that hit a gateway error or a dropped connection; waits honor Retry-After.

Explore the API

Type client. and your IDE lists what follows; hover over an operation, parameter, or field to read its documentation. In a REPL, help(client.document.v1.documents.list) prints the same for an operation. An operation's documentation ends with the scope it requires: an OAuth scope on PlatformClient and AccountClient, such as document:documents:read, and a token scope on ClassicClient and ClusterClient, such as logs.ingest.

Every path names its API version, as in client.document.v1.documents, client.storage.query.v1, and AccountClient().iam.v1.users, and where an API has several versions, each is a node of its own. PlatformClient reaches the classic environment APIs too, as client.classic.environment.v2 and .v1, with platform credentials; the docs use that only to send data and use ClassicClient for everything else classic.

Documentation

Features is a tour of the SDK. The tutorial starts with First steps and adds one idea per page; each page about an API shows most of its operations and ends with a table of all of them and the scopes they require.

• Basics: First steps, Authentication, API versions, Pagination, Settings
• Platform: DQL queries, Storage, Resource store, Documents, Automation, Anomaly detection, Smartscape, Problems, Davis CoPilot, Credential vault, Extensions, Fleet management, IAM, Synthetic, Vulnerabilities
• Classic: Metrics, Logs, Events, Topology, OpenTelemetry, ActiveGates, API tokens, Extensions, Network zones, OneAgents, RUM, SLOs, Synthetic, Unified analysis
• Account: Audit logs, IAM, Subscriptions
• Advanced: Concurrent operations, Rate limits, Client lifecycle, Object IDs, Exceptions, OAuth

The code on every page, and in this README, is type-checked and linted by tests/test_docs.py, which also checks that every print() shows a sample of its output.