- list_deployments - List Deployments
- get_deployment - Get Deployment
- get_deployment_logs - Get Deployment Logs
- stream_deployment_logs - Stream Deployment Logs
List Deployments
from mistralai.client import Mistral
import os
with Mistral(
api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:
res = mistral.workflows.deployments.list_deployments(active_only=True)
# Handle response
print(res)| Parameter | Type | Required | Description |
|---|---|---|---|
active_only |
Optional[bool] | ➖ | N/A |
is_hardened |
OptionalNullable[bool] | ➖ | Filter deployments by hardened status |
workflow_name |
OptionalNullable[str] | ➖ | N/A |
search |
OptionalNullable[str] | ➖ | Filter deployments by name or ID prefix |
limit |
OptionalNullable[int] | ➖ | Maximum number of deployments to return |
cursor |
OptionalNullable[str] | ➖ | Cursor from a previous response for pagination |
workspace_id |
OptionalNullable[str] | ➖ | Workspace ID to scope the request to. Defaults to the caller's context. |
retries |
Optional[utils.RetryConfig] | ➖ | Configuration to override the default retry behavior of the client. |
| Error Type | Status Code | Content Type |
|---|---|---|
| errors.HTTPValidationError | 422 | application/json |
| errors.SDKError | 4XX, 5XX | */* |
Get Deployment
from mistralai.client import Mistral
import os
with Mistral(
api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:
res = mistral.workflows.deployments.get_deployment(name="<value>")
# Handle response
print(res)| Parameter | Type | Required | Description |
|---|---|---|---|
name |
str | ✔️ | N/A |
retries |
Optional[utils.RetryConfig] | ➖ | Configuration to override the default retry behavior of the client. |
models.DeploymentDetailResponse
| Error Type | Status Code | Content Type |
|---|---|---|
| errors.HTTPValidationError | 422 | application/json |
| errors.SDKError | 4XX, 5XX | */* |
Retrieve logs for a deployment (across all of its workers).
Use after/before/order on the first request to set the time range and sort order; for
the next pages pass the cursor from the previous response (it remembers the range and order).
from mistralai.client import Mistral
import os
with Mistral(
api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:
res = mistral.workflows.deployments.get_deployment_logs(name="<value>", order="asc", limit=50)
# Handle response
print(res)| Parameter | Type | Required | Description |
|---|---|---|---|
name |
str | ✔️ | N/A |
worker_name |
OptionalNullable[str] | ➖ | Filter logs by worker name |
workflow_name |
OptionalNullable[str] | ➖ | Filter logs by workflow name |
after |
date | ➖ | Only return logs at or after this timestamp |
before |
date | ➖ | Only return logs before this timestamp |
order |
Optional[models.GetDeploymentLogsOrder] | ➖ | First-page sort order: 'asc' (oldest first) or 'desc'. Ignored when cursor is set. |
cursor |
OptionalNullable[str] | ➖ | Pagination cursor from a previous response's next_cursor; carries the window and order |
limit |
Optional[int] | ➖ | Maximum number of logs to return |
retries |
Optional[utils.RetryConfig] | ➖ | Configuration to override the default retry behavior of the client. |
models.DeploymentLogSearchResponse
| Error Type | Status Code | Content Type |
|---|---|---|
| errors.HTTPValidationError | 422 | application/json |
| errors.SDKError | 4XX, 5XX | */* |
Stream logs for a deployment (all of its workers) via SSE.
Resume cursor comes from the Last-Event-ID header or last_event_id query param (header wins)
and takes precedence over after; omit all to tail from the deployment start.
from mistralai.client import Mistral
import os
with Mistral(
api_key=os.getenv("MISTRAL_API_KEY", ""),
) as mistral:
res = mistral.workflows.deployments.stream_deployment_logs(name="<value>")
with res as event_stream:
for event in event_stream:
# handle event
print(event, flush=True)| Parameter | Type | Required | Description |
|---|---|---|---|
name |
str | ✔️ | N/A |
worker_name |
OptionalNullable[str] | ➖ | Filter logs by worker name |
workflow_name |
OptionalNullable[str] | ➖ | Filter logs by workflow name |
after |
date | ➖ | Start a fresh stream at this timestamp (ignored when resuming via last_event_id) |
last_event_id |
OptionalNullable[str] | ➖ | Resume from this cursor (a prior response's SSE id) |
retries |
Optional[utils.RetryConfig] | ➖ | Configuration to override the default retry behavior of the client. |
| Error Type | Status Code | Content Type |
|---|---|---|
| errors.HTTPValidationError | 422 | application/json |
| errors.SDKError | 4XX, 5XX | */* |