From 336962a8bebe826e72a8b96a43be17869176e038 Mon Sep 17 00:00:00 2001 From: Peter Kraft Date: Thu, 28 May 2026 11:42:27 -0700 Subject: [PATCH 1/2] update --- dbos_mcp/client.py | 26 ++++++++++++++++++++++++-- dbos_mcp/server.py | 45 ++++++++++++++++++++++++++++++++++++++++----- 2 files changed, 64 insertions(+), 7 deletions(-) diff --git a/dbos_mcp/client.py b/dbos_mcp/client.py index f5883ee..97adac1 100644 --- a/dbos_mcp/client.py +++ b/dbos_mcp/client.py @@ -499,14 +499,23 @@ async def get_workflow_aggregates( group_by_queue_name: bool = False, group_by_executor_id: bool = False, group_by_application_version: bool = False, + select_count: bool = False, + select_min_created_at: bool = False, + select_max_queue_wait_ms: bool = False, + select_max_total_latency_ms: bool = False, status: list[str] | None = None, start_time: str | None = None, end_time: str | None = None, + completed_after: str | None = None, + completed_before: str | None = None, + dequeued_after: str | None = None, + dequeued_before: str | None = None, name: list[str] | None = None, app_version: list[str] | None = None, executor_id: list[str] | None = None, queue_name: list[str] | None = None, workflow_id_prefix: list[str] | None = None, + time_bucket_size_ms: int | None = None, ) -> list[dict[str, Any]]: """Get workflow aggregates (counts grouped by dimensions).""" creds = _get_credentials() @@ -516,6 +525,10 @@ async def get_workflow_aggregates( "group_by_queue_name": group_by_queue_name, "group_by_executor_id": group_by_executor_id, "group_by_application_version": group_by_application_version, + "select_count": select_count, + "select_min_created_at": select_min_created_at, + "select_max_queue_wait_ms": select_max_queue_wait_ms, + "select_max_total_latency_ms": select_max_total_latency_ms, } if status is not None: body["status"] = status @@ -523,6 +536,14 @@ async def get_workflow_aggregates( body["start_time"] = start_time if end_time is not None: body["end_time"] = end_time + if completed_after is not None: + body["completed_after"] = completed_after + if completed_before is not None: + body["completed_before"] = completed_before + if dequeued_after is not None: + body["dequeued_after"] = dequeued_after + if dequeued_before is not None: + body["dequeued_before"] = dequeued_before if name is not None: body["name"] = name if app_version is not None: @@ -533,6 +554,8 @@ async def get_workflow_aggregates( body["queue_name"] = queue_name if workflow_id_prefix is not None: body["workflow_id_prefix"] = workflow_id_prefix + if time_bucket_size_ms is not None: + body["time_bucket_size_ms"] = time_bucket_size_ms async with httpx.AsyncClient() as client: response = await client.post( @@ -545,8 +568,7 @@ async def get_workflow_aggregates( timeout=30.0, ) response.raise_for_status() - data = response.json() - result: list[dict[str, Any]] = data.get("output", []) + result: list[dict[str, Any]] = response.json() return result diff --git a/dbos_mcp/server.py b/dbos_mcp/server.py index 3411fc4..cc4e812 100644 --- a/dbos_mcp/server.py +++ b/dbos_mcp/server.py @@ -567,20 +567,34 @@ async def get_workflow_aggregates( group_by_queue_name: bool = False, group_by_executor_id: bool = False, group_by_application_version: bool = False, + select_count: bool = False, + select_min_created_at: bool = False, + select_max_queue_wait_ms: bool = False, + select_max_total_latency_ms: bool = False, status: list[str] | None = None, start_time: str | None = None, end_time: str | None = None, + completed_after: str | None = None, + completed_before: str | None = None, + dequeued_after: str | None = None, + dequeued_before: str | None = None, name: list[str] | None = None, app_version: list[str] | None = None, executor_id: list[str] | None = None, queue_name: list[str] | None = None, workflow_id_prefix: list[str] | None = None, + time_bucket_size_ms: int | None = None, ) -> dict[str, Any]: - """Get workflow aggregate counts from DBOS Conductor. + """Get workflow aggregate metrics from DBOS Conductor. - Returns workflow counts grouped by one or more dimensions. Useful for + Returns workflow aggregates grouped by one or more dimensions. Useful for dashboards and understanding workflow status at a glance (e.g., "how many - workflows failed today?" or "how many workflows are pending per queue?"). + workflows failed today?", "how many workflows are pending per queue?", + "what's the worst queue-wait time per workflow name?"). + + Select at least one select_* flag to populate aggregate values. Select at + least one group_by_* flag to break the results down by dimension; with no + group_by_*, you get a single row across all matching workflows. Args: application_name (string, required): Name of the DBOS application @@ -589,19 +603,31 @@ async def get_workflow_aggregates( group_by_queue_name (bool, optional): Group results by queue name (default: false) group_by_executor_id (bool, optional): Group results by executor ID (default: false) group_by_application_version (bool, optional): Group results by application version (default: false) + select_count (bool, optional): Include count of workflows in each group (default: false) + select_min_created_at (bool, optional): Include min created_at (epoch ms) in each group (default: false) + select_max_queue_wait_ms (bool, optional): Include max queue wait time (ms) in each group (default: false) + select_max_total_latency_ms (bool, optional): Include max end-to-end latency (ms) in each group (default: false) status (array of strings, optional): Filter to these statuses before aggregating start_time (string, optional): Filter workflows created after this time (ISO 8601) end_time (string, optional): Filter workflows created before this time (ISO 8601) + completed_after (string, optional): Filter workflows completed after this time (ISO 8601) + completed_before (string, optional): Filter workflows completed before this time (ISO 8601) + dequeued_after (string, optional): Filter workflows dequeued after this time (ISO 8601) + dequeued_before (string, optional): Filter workflows dequeued before this time (ISO 8601) name (array of strings, optional): Filter to these workflow names before aggregating app_version (array of strings, optional): Filter to these application versions executor_id (array of strings, optional): Filter to these executor IDs queue_name (array of strings, optional): Filter to these queue names workflow_id_prefix (array of strings, optional): Filter to workflow IDs starting with these prefixes + time_bucket_size_ms (int, optional): Bucket aggregates into time windows of this many milliseconds Returns: aggregates: Array of aggregate objects, each containing: - - group (object): Map of dimension names to values (e.g., {"status": "ERROR", "workflow_name": "processOrder"}) - - count (int): Number of workflows matching this group + - group (object): Map of dimension names to values (e.g., {"status": "ERROR", "name": "processOrder"}) + - count (int, optional): Number of workflows matching this group (if select_count) + - min_created_at (int, optional): Earliest created_at in epoch ms (if select_min_created_at) + - max_queue_wait_ms (int, optional): Max queue wait time in ms (if select_max_queue_wait_ms) + - max_total_latency_ms (int, optional): Max end-to-end latency in ms (if select_max_total_latency_ms) application (string): Name of the application queried """ aggregates = await client.get_workflow_aggregates( @@ -611,14 +637,23 @@ async def get_workflow_aggregates( group_by_queue_name=group_by_queue_name, group_by_executor_id=group_by_executor_id, group_by_application_version=group_by_application_version, + select_count=select_count, + select_min_created_at=select_min_created_at, + select_max_queue_wait_ms=select_max_queue_wait_ms, + select_max_total_latency_ms=select_max_total_latency_ms, status=status, start_time=start_time, end_time=end_time, + completed_after=completed_after, + completed_before=completed_before, + dequeued_after=dequeued_after, + dequeued_before=dequeued_before, name=name, app_version=app_version, executor_id=executor_id, queue_name=queue_name, workflow_id_prefix=workflow_id_prefix, + time_bucket_size_ms=time_bucket_size_ms, ) return { "aggregates": aggregates, From 5b3e10b52b3812c39e21aabac1a35b2a6d01e486 Mon Sep 17 00:00:00 2001 From: Peter Kraft Date: Thu, 28 May 2026 11:50:26 -0700 Subject: [PATCH 2/2] update --- dbos_mcp/server.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/dbos_mcp/server.py b/dbos_mcp/server.py index cc4e812..ddb522d 100644 --- a/dbos_mcp/server.py +++ b/dbos_mcp/server.py @@ -592,9 +592,9 @@ async def get_workflow_aggregates( workflows failed today?", "how many workflows are pending per queue?", "what's the worst queue-wait time per workflow name?"). - Select at least one select_* flag to populate aggregate values. Select at - least one group_by_* flag to break the results down by dimension; with no - group_by_*, you get a single row across all matching workflows. + Select at least one select_* flag to populate aggregate values. At least + one group_by_* flag is mandatory to break the results down by dimension; + a query with no group_by_* will fail. Args: application_name (string, required): Name of the DBOS application