Query temporal directly for workflow runs #471
Conversation
|
Caution Review failedThe pull request is closed. Note Other AI code review bot(s) detectedCodeRabbit has detected other AI code review bot(s) in this pull request and will avoid duplicating their findings in the review comments. This may lead to a less comprehensive review. WalkthroughAdds paginated workflow-run listing (server → registry → Temporal) with base64 tokens, enriches statuses with workflow_id/run_id, introduces async server resolution in CLI workflow commands, adds XDG alternate credential paths and atomic credential writes, and renames the cloud CLI app namespace. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant CLI as CLI Client
participant App as App Server
participant Reg as Workflow Registry
participant Temp as Temporal
CLI->>App: list_workflow_runs(limit, page_size, next_page_token(b64))
App->>App: decode next_page_token (base64→bytes)
App->>Reg: list_workflow_statuses(query, limit, page_size, next_page_token=bytes)
alt Temporal-backed path
Reg->>Temp: list_workflows(query, page_size, next_page_token)
Temp-->>Reg: runs[], next_page_token?
Reg->>Temp: describe_workflow(run) (as needed)
Reg-->>App: WorkflowRunsPage{runs, next_page_token?}
else Local fallback
Reg->>Reg: aggregate local statuses (sort by updated_at)
Reg-->>App: runs[]
end
App-->>CLI: WorkflowRunsPage or runs[]
Note over CLI: If paged, present next_page_token (base64) to user
sequenceDiagram
autonumber
participant Cmd as CLI Workflow Cmd
participant Utils as resolve_server_async
participant Cloud as Cloud API Client
Cmd->>Utils: await resolve_server_async(id_or_url)
Utils->>Cloud: get_app_or_config(id_or_url) (await)
Cloud-->>Utils: MCPApp or MCPAppConfiguration
Utils-->>Cmd: resolved server
Cmd->>Cmd: extract serverUrl or raise CLIError
Cmd->>Cmd: determine API key (settings.API_KEY or stored credentials)
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested reviewers
Poem
Pre-merge checks and finishing touches✅ Passed checks (3 passed)
📜 Recent review detailsConfiguration used: CodeRabbit UI Review profile: CHILL Plan: Pro 📒 Files selected for processing (1)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (8)
src/mcp_agent/executor/workflow.py (1)
589-591: Status payload now includes workflow_id and run_id — looks good.Keeps backward compatibility via "id" while adding explicit identifiers. Consider documenting that "id" mirrors run_id and may be deprecated in favor of "run_id" to avoid ambiguity.
src/mcp_agent/server/app_server.py (3)
1047-1068: Docstring/typing should reflect union return (WorkflowRunsPage or List).The return section still says “A list...”. Update it to mention the WorkflowRunsPage envelope when a next_page_token exists, matching the type hint and registry contract.
- Returns: - A list of workflow run status dictionaries with detailed workflow information. + Returns: + List[Dict[str, Any]] when no further pages exist, or WorkflowRunsPage + when a next_page_token is present.
1081-1090: Harden Base64 decoding of next_page_token.Use validate=True and surface a client error rather than silently dropping invalid tokens.
- if next_page_token: - try: - import base64 as _b64 - - token_bytes = _b64.b64decode(next_page_token) - except Exception: - token_bytes = None + if next_page_token: + import base64 as _b64, binascii as _binascii + try: + token_bytes = _b64.b64decode(next_page_token, validate=True) + except (_binascii.Error, ValueError) as e: + raise ToolError(f"invalid next_page_token: {e}")
1092-1099: Pass-through looks correct; consider light input validation.Clamp negative/zero limit/page_size to sane values before forwarding to the registry to avoid backend surprises.
- await server_context.workflow_registry.list_workflow_statuses( + await server_context.workflow_registry.list_workflow_statuses( query=None, - limit=limit, - page_size=page_size, + limit=limit if isinstance(limit, int) and limit > 0 else 100, + page_size=page_size if (isinstance(page_size, int) and page_size > 0) else None, next_page_token=token_bytes, )src/mcp_agent/executor/temporal/workflow_registry.py (3)
224-228: Type mismatch vs abstract base: rpc_timeout should be timedelta | None.The abstract method uses timedelta; here it’s Any. Align to avoid LSP issues and mypy complaints.
- rpc_timeout: Any | None = None, + rpc_timeout: "timedelta | None" = None,
326-333: Access next_page_token defensively.Some SDK iterators may not expose next_page_token; use getattr to avoid AttributeError.
- if iterator.next_page_token: + _npt = getattr(iterator, "next_page_token", None) + if _npt: return WorkflowRunsPage( runs=results, - next_page_token=base64.b64encode(iterator.next_page_token).decode( + next_page_token=base64.b64encode(_npt).decode( "ascii" ), )
336-353: Apply limit on fallback path as well.Currently returns all local runs when Temporal listing fails, ignoring limit.
- # Fallback – return local cache augmented with Temporal describe where possible - for run_id, wf in self._local_workflows.items(): + # Fallback – return local cache augmented with Temporal describe where possible + count = 0 + max_count = limit if isinstance(limit, int) and limit > 0 else None + for run_id, wf in self._local_workflows.items(): status = await wf.get_status() workflow_id = wf.id or wf.name try: status["temporal"] = await self._get_temporal_workflow_status( workflow_id=workflow_id, run_id=run_id ) except Exception: # This is expected if we couldn't get a hold of the temporal client pass results.append(status) + count += 1 + if max_count is not None and count >= max_count: + breaksrc/mcp_agent/executor/workflow_registry.py (1)
24-27: WorkflowRunsPage model — LGTM.Simple and sufficient for transport. You may consider adding a brief docstring on token encoding expectations.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (4)
src/mcp_agent/executor/temporal/workflow_registry.py(3 hunks)src/mcp_agent/executor/workflow.py(1 hunks)src/mcp_agent/executor/workflow_registry.py(4 hunks)src/mcp_agent/server/app_server.py(3 hunks)
🧰 Additional context used
🧠 Learnings (1)
📚 Learning: 2025-09-05T14:31:48.139Z
Learnt from: rholinshead
PR: lastmile-ai/mcp-agent#414
File: src/mcp_agent/logging/logger.py:18-19
Timestamp: 2025-09-05T14:31:48.139Z
Learning: In the mcp-agent logging module (src/mcp_agent/logging/logger.py), temporalio should be imported lazily with try/except ImportError to avoid making it a hard dependency. Use temporalio.workflow.in_workflow() instead of isinstance checks on internal classes like _WorkflowInstanceImpl.
Applied to files:
src/mcp_agent/server/app_server.pysrc/mcp_agent/executor/temporal/workflow_registry.py
🧬 Code graph analysis (3)
src/mcp_agent/executor/workflow_registry.py (2)
src/mcp_agent/executor/temporal/workflow_registry.py (1)
list_workflow_statuses(219-352)src/mcp_agent/executor/workflow.py (1)
get_status(580-623)
src/mcp_agent/server/app_server.py (2)
src/mcp_agent/executor/workflow_registry.py (4)
WorkflowRegistry(29-178)WorkflowRunsPage(24-26)list_workflow_statuses(139-168)list_workflow_statuses(301-329)src/mcp_agent/executor/temporal/workflow_registry.py (1)
list_workflow_statuses(219-352)
src/mcp_agent/executor/temporal/workflow_registry.py (4)
src/mcp_agent/executor/workflow.py (4)
executor(118-123)run_id(133-138)id(126-130)get_status(580-623)src/mcp_agent/executor/workflow_registry.py (4)
WorkflowRegistry(29-178)WorkflowRunsPage(24-26)list_workflow_statuses(139-168)list_workflow_statuses(301-329)src/mcp_agent/executor/temporal/__init__.py (1)
ensure_client(260-275)src/mcp_agent/logging/logger.py (1)
warning(281-289)
🔇 Additional comments (5)
src/mcp_agent/server/app_server.py (1)
26-28: Importing WorkflowRunsPage for union return type — OK.src/mcp_agent/executor/temporal/workflow_registry.py (2)
12-12: Using public WorkflowRunsPage — good alignment with registry API.
257-264: No SDK signature mismatch — keep the call as-is.
Verified that Temporal Python SDK's Client.list_workflows accepts query, limit, page_size, next_page_token, rpc_metadata, and rpc_timeout; the introspection fallback is unnecessary.src/mcp_agent/executor/workflow_registry.py (2)
139-148: Expanded list_workflow_statuses signature — good; aligns with server and Temporal impl.Ensure concrete implementations keep the rpc_timeout type as timedelta | None for consistency (Temporal impl currently uses Any | None).
301-329: In-memory listing behavior — OK.Recency sort + optional limit is fine; ignoring pagination tokens is acceptable per docstring.
| # Use caller query if provided; else default to newest first | ||
| query_local = query or "order by StartTime desc" | ||
|
|
||
| iterator = client.list_workflows( |
There was a problem hiding this comment.
Just noting for later -- when we push support for multiple users using a server, we'll need to add the auth checks in the proxy layer for this request, which will have implications for results returned (e.g. we'll only return those that the user has permission to see)
There was a problem hiding this comment.
Good point, let me mark that as a TODO comment in the code as well. I think this is reasonable for the time being but agreed we'll want this to be user-scoped in the future.
There was a problem hiding this comment.
Fwiw, this client code shouldn't really need to know/care about it, as long as our rpc metadata passes through the api key. Our backend proxy layer will handle the auth checks, but will need to handle pagination and such with some in-memory processing
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (8)
src/mcp_agent/cli/mcp_app/mcp_client.py (1)
181-201: Robust pagination parsing; avoid silent drops and JSON round‑trips
- Swallowing all exceptions while parsing runs can hide data issues; at least collect a count for diagnostics.
- Minor perf nit: avoid dict→JSON→dict round‑trip when item is already a dict.
Apply this diff to remove the JSON round‑trip at call sites:
- runs.append( - MCPClientSession.deserialize_workflow_run(json.dumps(r)) - ) + runs.append(MCPClientSession.deserialize_workflow_run_obj(r))- runs.append( - MCPClientSession.deserialize_workflow_run(json.dumps(r)) - ) + runs.append(MCPClientSession.deserialize_workflow_run_obj(r))- runs.append( - MCPClientSession.deserialize_workflow_run(json.dumps(data)) - ) + runs.append(MCPClientSession.deserialize_workflow_run_obj(data))And add this helper to handle dicts (outside the changed hunk):
@staticmethod def deserialize_workflow_run_obj(obj: dict[str, Any]) -> WorkflowRun: # Mirror string path’s result parsing for "result" if needed run_data = dict(obj) if isinstance(run_data.get("result"), str): try: run_data["result"] = ast.literal_eval(run_data["result"]) except (ValueError, SyntaxError): try: run_data["result"] = json.loads(run_data["result"]) except json.JSONDecodeError: # Leave as-is; WorkflowRun will validate pass return WorkflowRun(**run_data)Optionally, tally parse failures and include a debug log or return a warning counter in the client result.
Also applies to: 210-271
src/mcp_agent/server/app_server.py (3)
1057-1063: Paginated tool signature — good; align docstring “Returns”The return type can be a WorkflowRunsPage; docstring still says list only.
Apply this docstring adjustment:
- Returns: - A list of workflow run status dictionaries with detailed workflow information. + Returns: + List[Dict[str, Any]] | WorkflowRunsPage: + A list of workflow run status dictionaries, or a page object when pagination continues.
1091-1100: Invalid next_page_token handlingSilently nulling invalid base64 can yield confusing paging loops. Prefer a 400-style ToolError.
- token_bytes = None - if next_page_token: - try: - import base64 as _b64 - - token_bytes = _b64.b64decode(next_page_token) - except Exception: - token_bytes = None + token_bytes = None + if next_page_token: + try: + import base64 as _b64 + token_bytes = _b64.b64decode(next_page_token, validate=True) + except Exception: + raise ToolError("Invalid next_page_token (must be base64-encoded)") # 400-equivalent
1102-1109: Guardrails on limit/page_sizeConsider clamping to a sane max (e.g., 1000) to avoid heavy backend scans.
- workflow_statuses = ( - await server_context.workflow_registry.list_workflow_statuses( - query=None, - limit=limit, - page_size=page_size, - next_page_token=token_bytes, - ) - ) + # Simple bounds to protect backends + eff_limit = max(1, min(limit or 100, 1000)) + eff_page = None if page_size is None else max(1, min(page_size, 1000)) + workflow_statuses = await server_context.workflow_registry.list_workflow_statuses( + query=None, + limit=eff_limit, + page_size=eff_page, + next_page_token=token_bytes, + )src/mcp_agent/executor/temporal/workflow_registry.py (4)
255-270: Compat path for older Temporal SDKsSome SDK versions may not accept rpc_metadata/rpc_timeout on list_workflows. Fall back to a minimal call instead of bailing to local cache.
- iterator = client.list_workflows( - query=query_local, - limit=limit, - page_size=page_size, - next_page_token=next_page_token, - rpc_metadata=rpc_metadata, - rpc_timeout=rpc_timeout, - ) + try: + iterator = client.list_workflows( + query=query_local, + limit=limit, + page_size=page_size, + next_page_token=next_page_token, + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, + ) + except TypeError: + # Fallback for older SDKs without rpc_* kwargs + iterator = client.list_workflows( + query=query_local, + limit=limit, + page_size=page_size, + next_page_token=next_page_token, + )
350-360: Running-state detectionTemporal status “OPEN” isn’t a canonical execution status in newer SDKs; using only RUNNING is safer.
- status_dict["running"] = ts.upper() in {"RUNNING", "OPEN"} + status_dict["running"] = ts.upper() == "RUNNING"
279-287: Double limitingYou pass limit to the backend and enforce locally. That’s fine, but if the backend already respects limit, the local cap is redundant. Optional cleanup.
329-342: Temporal payload naming consistencyIn the temporal blob, “name” holds the workflow_id; consider using workflow_type for display (“type” is already present) or drop “name” to avoid confusion.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
src/mcp_agent/cli/core/utils.py(1 hunks)src/mcp_agent/cli/mcp_app/mcp_client.py(4 hunks)src/mcp_agent/executor/temporal/workflow_registry.py(3 hunks)src/mcp_agent/server/app_server.py(11 hunks)
🧰 Additional context used
🧠 Learnings (1)
📚 Learning: 2025-09-05T14:31:48.139Z
Learnt from: rholinshead
PR: lastmile-ai/mcp-agent#414
File: src/mcp_agent/logging/logger.py:18-19
Timestamp: 2025-09-05T14:31:48.139Z
Learning: In the mcp-agent logging module (src/mcp_agent/logging/logger.py), temporalio should be imported lazily with try/except ImportError to avoid making it a hard dependency. Use temporalio.workflow.in_workflow() instead of isinstance checks on internal classes like _WorkflowInstanceImpl.
Applied to files:
src/mcp_agent/executor/temporal/workflow_registry.pysrc/mcp_agent/server/app_server.py
🧬 Code graph analysis (4)
src/mcp_agent/cli/mcp_app/mcp_client.py (2)
src/mcp_agent/executor/workflow_registry.py (1)
WorkflowRunsPage(24-26)src/mcp_agent/cli/cloud/commands/workflows/runs/main.py (1)
list_workflow_runs(87-115)
src/mcp_agent/executor/temporal/workflow_registry.py (3)
src/mcp_agent/executor/workflow.py (4)
executor(118-123)run_id(133-138)id(126-130)get_status(580-623)src/mcp_agent/executor/workflow_registry.py (3)
WorkflowRunsPage(24-26)list_workflow_statuses(139-168)list_workflow_statuses(301-329)src/mcp_agent/executor/temporal/__init__.py (1)
ensure_client(260-275)
src/mcp_agent/cli/core/utils.py (2)
src/mcp_agent/config.py (1)
Settings(583-690)src/mcp_agent/cli/mcp_app/api_client.py (1)
MCPApp(22-29)
src/mcp_agent/server/app_server.py (2)
src/mcp_agent/executor/workflow_registry.py (4)
WorkflowRegistry(29-178)WorkflowRunsPage(24-26)list_workflow_statuses(139-168)list_workflow_statuses(301-329)src/mcp_agent/executor/temporal/workflow_registry.py (1)
list_workflow_statuses(220-394)
🔇 Additional comments (6)
src/mcp_agent/cli/core/utils.py (1)
30-32: Signature reformat only — LGTMNo behavioral change; reads cleaner.
src/mcp_agent/server/app_server.py (2)
26-28: Importing WorkflowRunsPage into server surface — LGTM
531-533: Non-functional formatting/serialization tweaksNo action needed.
Also applies to: 609-611, 633-635, 652-654, 671-673, 869-869
src/mcp_agent/executor/temporal/workflow_registry.py (2)
220-229: API surface update — LGTMKeyword-only params and union return type fit the new paging contract.
368-375: Page token encoding — LGTMOnly returning WorkflowRunsPage when a next_page_token exists matches the contract.
src/mcp_agent/cli/mcp_app/mcp_client.py (1)
146-147: Expose next_page_token in the CLI so users can paginateSearch found callers of list_workflow_runs but none appear to surface or consume the new next_page_token — wire the token through the client → CLI (or add a --page-token / “load more” flow) so users can continue pagination.
- src/mcp_agent/cli/mcp_app/mcp_client.py (list_workflow_runs)
- src/mcp_agent/cli/cloud/commands/app/workflows/main.py (res = await session.list_workflow_runs())
- src/mcp_agent/cli/cloud/commands/workflows/runs/main.py (result = await mcp_client_session.list_workflow_runs())
- src/mcp_agent/server/app_server.py (tool: list_workflow_runs)
![graphite-app[bot]](https://avatars.githubusercontent.com/in/158384?s=60&v=4){kind=small}
| next_page_token=base64.b64encode(iterator.next_page_token).decode( | ||
| "ascii" | ||
| ), |
There was a problem hiding this comment.
Potential encoding/decoding mismatch: The code base64-encodes iterator.next_page_token assuming it's bytes, but later in the server code (lines 1095-1097) it's decoded back to bytes. If iterator.next_page_token is already a string, this will cause a TypeError on b64encode(). The code should verify the type of next_page_token before encoding or handle both string and bytes cases.
| next_page_token=base64.b64encode(iterator.next_page_token).decode( | |
| "ascii" | |
| ), | |
| next_page_token=base64.b64encode( | |
| iterator.next_page_token.encode('utf-8') if isinstance(iterator.next_page_token, str) else iterator.next_page_token | |
| ).decode("ascii"), |
Spotted by Diamond
Is this helpful? React 👍 or 👎 to let us know.
| def _to_timestamp(dt: datetime): | ||
| try: | ||
| return dt.timestamp() if dt is not None else None | ||
| except Exception: | ||
| return dt |
There was a problem hiding this comment.
Unsafe exception handling in timestamp conversion: The _to_timestamp() function catches all exceptions and returns the original datetime object on error, but this could cause type inconsistency issues downstream where code expects a numeric timestamp. If dt.timestamp() fails, returning the datetime object could cause runtime errors in code that expects a number. Fix: Return None or a default numeric value on conversion failure.
| def _to_timestamp(dt: datetime): | |
| try: | |
| return dt.timestamp() if dt is not None else None | |
| except Exception: | |
| return dt | |
| def _to_timestamp(dt: datetime): | |
| if dt is None: | |
| return None | |
| try: | |
| return dt.timestamp() | |
| except Exception: | |
| return None |
Spotted by Diamond
Is this helpful? React 👍 or 👎 to let us know.
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
src/mcp_agent/cli/cloud/main.py (1)
86-91: 'apps list' is wired to servers list; add proper impl or remove to prevent confusion.
apps listcurrently invokesservers.list_servers, which is misleading. Either implementapps listor drop the command until ready.-app_cmd_app.command(name="list")(list_servers) +# TODO: Implement `apps list`. Removing miswired command to avoid confusion. +# app_cmd_app.command(name="list")(list_servers)src/mcp_agent/cli/cloud/commands/workflows/runs/main.py (1)
57-71: Only the first page is fetched; implement pagination to honor --limit and server paging.Temporal/registry now paginate; without looping you’ll miss older runs and
--limit> page size is not honored.- with console.status( - "[bold green]Fetching workflow runs...", spinner="dots" - ): - result = await mcp_client_session.list_workflow_runs() - - workflows = ( - result.workflow_runs if result and result.workflow_runs else [] - ) - - if status: - workflows = [w for w in workflows if _matches_status(w, status)] - - if limit: - workflows = workflows[:limit] + with console.status( + "[bold green]Fetching workflow runs...", spinner="dots" + ): + # Page until we satisfy `limit` or exhaust results + page_token = None + page_size = min(limit, 100) if limit else 100 + collected = [] + while True: + try: + kwargs = {"page_size": page_size} + if page_token: + kwargs["page_token"] = page_token + result = await mcp_client_session.list_workflow_runs(**kwargs) + except TypeError: + # Backward-compat: server may not support pagination + result = await mcp_client_session.list_workflow_runs() + + page = result.workflow_runs if result and result.workflow_runs else [] + collected.extend(page) + + if limit and len(collected) >= limit: + break + next_token = getattr(result, "next_page_token", None) or getattr( + result, "nextPageToken", None + ) + if not next_token: + break + page_token = next_token + + workflows = collected[:limit] if limit else collected + + if status: + workflows = [w for w in workflows if _matches_status(w, status)]src/mcp_agent/executor/temporal/workflow_registry.py (1)
186-203: get_workflow_status returns False on error but type is Optional[Dict]; return None.Returning False violates the declared return type and can break callers expecting a dict or None.
- return False + return None @@ - return False + return None
♻️ Duplicate comments (2)
src/mcp_agent/executor/temporal/workflow_registry.py (2)
318-323: Return None on timestamp conversion failure (avoid type drift).Returning the datetime object on failure causes type inconsistency downstream. Prefer None.
- def _to_timestamp(dt: datetime): - try: - return dt.timestamp() if dt is not None else None - except Exception: - return dt + def _to_timestamp(dt: datetime): + if dt is None: + return None + try: + return dt.timestamp() + except Exception: + return None
365-371: Base64-encode next_page_token safely for str/bytes.iterator.next_page_token may be str or bytes; handle both to avoid TypeError.
- if iterator.next_page_token: + if iterator.next_page_token: return WorkflowRunsPage( runs=results, - next_page_token=base64.b64encode(iterator.next_page_token).decode( - "ascii" - ), + next_page_token=base64.b64encode( + iterator.next_page_token.encode("utf-8") + if isinstance(iterator.next_page_token, str) + else iterator.next_page_token + ).decode("ascii"), )
🧹 Nitpick comments (13)
src/mcp_agent/cli/auth/constants.py (1)
9-12: Normalize XDG path handling to avoid surprises when XDG_CONFIG_HOME contains env vars or "~".Expand env vars and "~" and pre-expand the joined path to ensure consistent behavior across shells/OS.
-# Additional locations to search (XDG-compatible and documented path) -XDG_CONFIG_HOME = os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser("~/.config") -ALTERNATE_CREDENTIALS_PATHS = [ - os.path.join(XDG_CONFIG_HOME, "mcp-agent", "credentials.json"), -] +# Additional locations to search (XDG-compatible and documented path) +XDG_CONFIG_HOME = os.path.expanduser( + os.path.expandvars(os.environ.get("XDG_CONFIG_HOME", "~/.config")) +) +ALTERNATE_CREDENTIALS_PATHS = [ + os.path.expanduser(os.path.join(XDG_CONFIG_HOME, "mcp-agent", "credentials.json")), +]src/mcp_agent/cli/cloud/commands/workflows/cancel/main.py (1)
43-45: Align auth error with new env‑key path.Message should mention MCP_API_KEY like other commands.
- if not effective_api_key: - raise CLIError("Must be logged in to access server. Run 'mcp-agent login'.") + if not effective_api_key: + raise CLIError("Not authenticated. Set MCP_API_KEY or run 'mcp-agent login'.")src/mcp_agent/cli/cloud/commands/workflows/runs/main.py (1)
49-51: Align auth error with env‑key path.Match the new pattern used elsewhere.
- if not effective_api_key: - raise CLIError("Must be logged in to access server. Run 'mcp-agent login'.") + if not effective_api_key: + raise CLIError("Not authenticated. Set MCP_API_KEY or run 'mcp-agent login'.")src/mcp_agent/cli/cloud/commands/workflows/resume/main.py (2)
27-31: Harden URL detection (use urlparse).startswith("http") misses uppercase schemes and future variants. Use urlparse for robustness.
Apply this diff:
+from urllib.parse import urlparse @@ - if server_id_or_url.startswith(("http://", "https://")): - server_url = server_id_or_url + parsed = urlparse(server_id_or_url) + if parsed.scheme in ("http", "https"): + server_url = server_id_or_url
66-68: Confirmed: client accepts arbitrary signal names — rename optional. mcp_client.resume_workflow(run_id, signal_name='resume', payload) accepts a signal_name argument (see src/mcp_agent/cli/mcp_app/mcp_client.py) and is used by the CLI; server has resume_workflow (src/mcp_agent/server/app_server.py) and the Temporal layer implements a generic signal_workflow, so no functional change needed — consider renaming to signal_workflow for clarity.src/mcp_agent/cli/cloud/commands/workflows/describe/main.py (3)
92-96: Reuse shared format validator for consistency.Use validate_output_format from commands.utils instead of manual check.
Apply:
-from ...utils import ( - handle_server_api_errors, - resolve_server_async, - setup_authenticated_client, -) +from ...utils import ( + handle_server_api_errors, + resolve_server_async, + setup_authenticated_client, + validate_output_format, +) @@ - if format not in ["text", "json", "yaml"]: - console.print("[red]Error: --format must be 'text', 'json', or 'yaml'[/red]") - raise typer.Exit(6) + validate_output_format(format)
124-126: Use UTC when formatting epoch timestamps.fromtimestamp() uses local time. Prefer explicit UTC to avoid confusion.
Apply:
-from datetime import datetime +from datetime import datetime, timezone @@ - created_dt = datetime.fromtimestamp(created_at) - created_at = created_dt.strftime("%Y-%m-%d %H:%M:%S") + created_dt = datetime.fromtimestamp(created_at, tz=timezone.utc) + created_at = created_dt.strftime("%Y-%m-%d %H:%M:%S UTC") @@ - start_dt = datetime.fromtimestamp(start_time).strftime( - "%Y-%m-%d %H:%M:%S" - ) + start_dt = datetime.fromtimestamp(start_time, tz=timezone.utc).strftime( + "%Y-%m-%d %H:%M:%S UTC" + ) @@ - end_dt = datetime.fromtimestamp(end_time).strftime("%Y-%m-%d %H:%M:%S") + end_dt = datetime.fromtimestamp(end_time, tz=timezone.utc).strftime( + "%Y-%m-%d %H:%M:%S UTC" + )Also applies to: 158-164
25-26: Nit: avoid shadowing built-in name “format”.Rename parameter to output_format for clarity.
Apply:
-async def _describe_workflow_async( - server_id_or_url: str, run_id: str, format: str = "text" +async def _describe_workflow_async( + server_id_or_url: str, run_id: str, output_format: str = "text" ): @@ - print_workflow_status(workflow_status, format) + print_workflow_status(workflow_status, output_format) @@ -def print_workflow_status(workflow_status: WorkflowRun, format: str = "text") -> None: +def print_workflow_status(workflow_status: WorkflowRun, output_format: str = "text") -> None: @@ - if format == "json": + if output_format == "json": @@ - elif format == "yaml": + elif output_format == "yaml": @@ - run_async(_describe_workflow_async(server_id_or_url, run_id, format)) + run_async(_describe_workflow_async(server_id_or_url, run_id, format))Note: final line keeps CLI option name “--format” intact; only internal name changes.
Also applies to: 96-96, 99-106
src/mcp_agent/cli/cloud/commands/utils.py (1)
39-53: Expose and encourage shared helpers.validate_output_format and resolve_server (sync wrapper) provide consistent ergonomics. Recommend updating callers (e.g., describe) to use them.
I can push a follow-up patch wiring these into all workflow commands if desired.
Also applies to: 76-85
src/mcp_agent/cli/auth/main.py (1)
34-41: Specify UTF‑8 when writing JSON.Read path uses encoding="utf-8"; align writer to avoid locale issues.
Apply:
- with os.fdopen(tmp_fd, "w") as f: + with os.fdopen(tmp_fd, "w", encoding="utf-8") as f:src/mcp_agent/server/app_server.py (2)
1057-1078: Return shape/doc mismatch; document paginated form explicitly.The tool can now return a paginated object. Update the Returns section to describe both shapes so clients aren’t surprised.
Returns: - A list of workflow run status dictionaries with detailed workflow information. + - When no pagination token is present: List[Dict[str, Any]] of run status entries. + - When pagination is supported and there is a token: Dict with shape + {"runs": List[Dict[str, Any]], "next_page_token": "<base64>"}.
1091-1100: Don’t silently swallow invalid next_page_token.If Base64 decode fails we drop the token without any signal. Log a warning so callers can diagnose paging issues; keep behavior non-fatal.
token_bytes = None if next_page_token: try: import base64 as _b64 token_bytes = _b64.b64decode(next_page_token) except Exception: - token_bytes = None + token_bytes = None + try: + logger.warning("Invalid next_page_token provided; ignoring.") + except Exception: + passsrc/mcp_agent/executor/temporal/workflow_registry.py (1)
326-333: Normalize temporal payload IDs and names (consistency with top-level).Top-level uses id=run_id; temporal block sets id=workflow_id and name=workflow_info.id. Align id to run_id and prefer workflow_type (human-friendly) for name.
- temporal_status = { - "id": workflow_id, + temporal_status = { + "id": run_id, "workflow_id": workflow_id, "run_id": run_id, - "name": workflow_info.id, - "type": workflow_type, + "name": workflow_type or workflow_id, + "type": workflow_type,
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (14)
examples/workflows/workflow_evaluator_optimizer/main.py(1 hunks)src/mcp_agent/cli/auth/constants.py(1 hunks)src/mcp_agent/cli/auth/main.py(4 hunks)src/mcp_agent/cli/cloud/commands/auth/whoami/main.py(2 hunks)src/mcp_agent/cli/cloud/commands/logger/tail/main.py(2 hunks)src/mcp_agent/cli/cloud/commands/utils.py(4 hunks)src/mcp_agent/cli/cloud/commands/workflows/cancel/main.py(3 hunks)src/mcp_agent/cli/cloud/commands/workflows/describe/main.py(3 hunks)src/mcp_agent/cli/cloud/commands/workflows/list/main.py(3 hunks)src/mcp_agent/cli/cloud/commands/workflows/resume/main.py(3 hunks)src/mcp_agent/cli/cloud/commands/workflows/runs/main.py(3 hunks)src/mcp_agent/cli/cloud/main.py(1 hunks)src/mcp_agent/executor/temporal/workflow_registry.py(3 hunks)src/mcp_agent/server/app_server.py(3 hunks)
✅ Files skipped from review due to trivial changes (1)
- examples/workflows/workflow_evaluator_optimizer/main.py
🧰 Additional context used
🧠 Learnings (1)
📚 Learning: 2025-09-05T14:31:48.139Z
Learnt from: rholinshead
PR: lastmile-ai/mcp-agent#414
File: src/mcp_agent/logging/logger.py:18-19
Timestamp: 2025-09-05T14:31:48.139Z
Learning: In the mcp-agent logging module (src/mcp_agent/logging/logger.py), temporalio should be imported lazily with try/except ImportError to avoid making it a hard dependency. Use temporalio.workflow.in_workflow() instead of isinstance checks on internal classes like _WorkflowInstanceImpl.
Applied to files:
src/mcp_agent/server/app_server.pysrc/mcp_agent/executor/temporal/workflow_registry.py
🧬 Code graph analysis (11)
src/mcp_agent/cli/cloud/commands/logger/tail/main.py (2)
src/mcp_agent/cli/auth/models.py (1)
UserCredentials(10-64)src/mcp_agent/cli/utils/ux.py (1)
print_error(80-91)
src/mcp_agent/cli/cloud/commands/workflows/resume/main.py (2)
src/mcp_agent/cli/cloud/commands/utils.py (1)
resolve_server_async(55-73)src/mcp_agent/cli/auth/main.py (1)
load_api_key_credentials(108-115)
src/mcp_agent/cli/cloud/commands/utils.py (4)
src/mcp_agent/cli/auth/main.py (1)
load_api_key_credentials(108-115)src/mcp_agent/cli/exceptions.py (1)
CLIError(4-9)src/mcp_agent/cli/mcp_app/api_client.py (3)
MCPApp(22-29)MCPAppConfiguration(33-38)get_app_or_config(247-283)src/mcp_agent/cli/core/utils.py (1)
run_async(12-27)
src/mcp_agent/cli/cloud/commands/workflows/cancel/main.py (2)
src/mcp_agent/cli/cloud/commands/utils.py (1)
resolve_server_async(55-73)src/mcp_agent/cli/auth/main.py (1)
load_api_key_credentials(108-115)
src/mcp_agent/cli/cloud/commands/auth/whoami/main.py (3)
src/mcp_agent/cli/auth/main.py (1)
load_credentials(59-85)src/mcp_agent/cli/auth/models.py (1)
UserCredentials(10-64)src/mcp_agent/cli/exceptions.py (1)
CLIError(4-9)
src/mcp_agent/cli/auth/main.py (2)
src/mcp_agent/cli/utils/ux.py (1)
print_warning(66-77)src/mcp_agent/cli/auth/models.py (3)
to_json(56-58)UserCredentials(10-64)from_json(61-64)
src/mcp_agent/server/app_server.py (2)
src/mcp_agent/executor/workflow_registry.py (4)
WorkflowRegistry(29-178)WorkflowRunsPage(24-26)list_workflow_statuses(139-168)list_workflow_statuses(301-329)src/mcp_agent/executor/temporal/workflow_registry.py (1)
list_workflow_statuses(220-391)
src/mcp_agent/cli/cloud/commands/workflows/describe/main.py (2)
src/mcp_agent/cli/cloud/commands/utils.py (1)
resolve_server_async(55-73)src/mcp_agent/cli/auth/main.py (1)
load_api_key_credentials(108-115)
src/mcp_agent/cli/cloud/commands/workflows/runs/main.py (2)
src/mcp_agent/cli/cloud/commands/utils.py (1)
resolve_server_async(55-73)src/mcp_agent/cli/auth/main.py (1)
load_api_key_credentials(108-115)
src/mcp_agent/executor/temporal/workflow_registry.py (3)
src/mcp_agent/executor/workflow.py (4)
executor(118-123)run_id(133-138)id(126-130)get_status(580-623)src/mcp_agent/executor/workflow_registry.py (6)
WorkflowRegistry(29-178)WorkflowRunsPage(24-26)list_workflow_statuses(139-168)list_workflow_statuses(301-329)list_workflows(171-178)list_workflows(331-332)src/mcp_agent/executor/temporal/__init__.py (1)
ensure_client(279-294)
src/mcp_agent/cli/cloud/commands/workflows/list/main.py (2)
src/mcp_agent/cli/cloud/commands/utils.py (1)
resolve_server_async(55-73)src/mcp_agent/cli/auth/main.py (1)
load_api_key_credentials(108-115)
🔇 Additional comments (14)
src/mcp_agent/cli/cloud/commands/logger/tail/main.py (1)
106-114: Env-based API key fallback verified — setup_authenticated_client honors MCP_API_KEYsetup_authenticated_client uses settings.API_KEY or load_api_key_credentials(), so the HTTP client path will use the same env-var fallback (see src/mcp_agent/cli/cloud/commands/utils.py around effective_api_key).
src/mcp_agent/cli/cloud/commands/workflows/cancel/main.py (1)
50-61: Don't pass --reason yet — cancel_workflow API doesn't accept areasonparameterSearch shows cancel_workflow signatures accept only run_id/workflow_id (no reason), so calling with reason will raise TypeError. Update the API before changing the CLI.
- Files to change to add optional reason:
- src/mcp_agent/cli/mcp_app/mcp_client.py (async def cancel_workflow(...))
- src/mcp_agent/cli/cloud/commands/workflows/cancel/main.py (call site)
- src/mcp_agent/server/app_server.py (tool
workflows-cancelsignature and call to registry)- src/mcp_agent/executor/workflow_registry.py (abstract signature)
- src/mcp_agent/executor/temporal/workflow_registry.py (implementation and cancellation logic)
Options: add an optional reason: str | None = None to the above signatures and propagate it; or leave the CLI behavior as-is (print reason only). Do not apply the suggested try/except change without first adding server/client support.
Likely an incorrect or invalid review comment.
src/mcp_agent/cli/cloud/commands/auth/whoami/main.py (2)
19-29: Good UX: env-based auth fallback and clearer not-authenticated message.The Panel notice and updated error text reduce confusion.
Also applies to: 31-33
7-8: No action required — re-exports present. src/mcp_agent/cli/auth/init.py re-exports load_credentials and UserCredentials; leave the import as-is.src/mcp_agent/cli/cloud/commands/workflows/describe/main.py (1)
32-33: Async server resolution and API key fallback look correct.Consistent with utils.resolve_server_async and env-first credential strategy.
Also applies to: 44-47
src/mcp_agent/cli/cloud/commands/utils.py (2)
28-36: Auth setup: env-first strategy LGTM.Clear error text and fallback to stored credentials are consistent across commands.
55-73: resolve_server_async wrapper is correct and safe.Good centralized error message and async usage.
src/mcp_agent/cli/auth/main.py (2)
28-56: Great: atomic, permissioned credential writes.mkstemp + fsync + os.replace with 0600 perms is solid.
65-85: Multi-path load and clear behaviors look good.Graceful handling of corrupted files with warnings; clearing across alternate paths is correct.
Also applies to: 94-105
src/mcp_agent/cli/cloud/commands/workflows/resume/main.py (1)
31-41: Async server resolution + env-key fallback verified.rg output shows resolve_server_async and the settings.API_KEY or load_api_key_credentials() fallback present across the cloud CLI commands (workflows, deploy, configure, apps); no changes required.
src/mcp_agent/cli/cloud/commands/workflows/list/main.py (3)
17-18: Async resolver swap looks good.Switching to resolve_server_async aligns with upstream client changes.
29-36: Server resolution flow is correct.Awaiting the async resolver and validating appServerInfo/serverUrl is sound.
If there are configs without appServerInfo but with a direct serverUrl field, confirm they are still handled elsewhere. If needed, we can add a fallback to getattr(server, "serverUrl", None).
41-44: Settings-first API key fallback is correct.Using settings.API_KEY before the legacy credentials file matches the new auth pattern.
src/mcp_agent/executor/temporal/workflow_registry.py (1)
279-359: N+1 describe avoided — good. Consider guarded enrichment only when needed.You’re already using fields from list_workflows and only describing on failure. That addresses the hot-path N+1. Keep it that way; if you later add a “details” knob, only call describe when details=True.
Add a load test that lists 5k+ executions and asserts a single List RPC and zero Describe RPCs in logs unless details=True.
| app_cmd_app.command(name="status")(get_app_status) | ||
| app_cmd_app.command(name="workflows")(list_app_workflows) | ||
| app.add_typer(app_cmd_app, name="app", help="Manage an MCP App") | ||
| app.add_typer(app_cmd_app, name="apps", help="Manage an MCP App") |
There was a problem hiding this comment.
Keep a backward‑compatible alias for the CLI rename app → apps.
This is a breaking change for users/scripts. Add a deprecated alias to smooth the transition.
-app.add_typer(app_cmd_app, name="apps", help="Manage an MCP App")
+app.add_typer(app_cmd_app, name="apps", help="Manage MCP Apps")
+# Back-compat alias (deprecated)
+app.add_typer(app_cmd_app, name="app", help="[DEPRECATED] Use 'apps'")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| app.add_typer(app_cmd_app, name="apps", help="Manage an MCP App") | |
| app.add_typer(app_cmd_app, name="apps", help="Manage MCP Apps") | |
| # Back-compat alias (deprecated) | |
| app.add_typer(app_cmd_app, name="app", help="[DEPRECATED] Use 'apps'") |
🤖 Prompt for AI Agents
In src/mcp_agent/cli/cloud/main.py around line 90, add a backward-compatible
deprecated alias for the renamed CLI group by registering the same Typer under
the old name "app" in addition to "apps". Call app.add_typer(app_cmd_app,
name="app", help="(deprecated) Manage an MCP App — use 'apps' instead") so
scripts using `app` continue to work; optionally add a small wrapper (or a Click
hook) that emits a deprecation warning when the `app` alias is invoked to inform
users to migrate to `apps`.
| ) | ||
| ) | ||
|
|
||
| return workflow_statuses |
There was a problem hiding this comment.
Ensure JSON-serializable return (Pydantic to dict).
FastMCP tools should return JSON-serializable data. If the registry returns a WorkflowRunsPage (Pydantic), convert it to a dict before returning.
- return workflow_statuses
+ if isinstance(workflow_statuses, WorkflowRunsPage):
+ return workflow_statuses.model_dump(by_alias=True, mode="json", exclude_none=True)
+ return workflow_statuses📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| return workflow_statuses | |
| if isinstance(workflow_statuses, WorkflowRunsPage): | |
| return workflow_statuses.model_dump(by_alias=True, mode="json", exclude_none=True) | |
| return workflow_statuses |
🤖 Prompt for AI Agents
In src/mcp_agent/server/app_server.py around line 1111, the function currently
returns a Pydantic WorkflowRunsPage object (not JSON-serializable); convert the
Pydantic model to a plain dict before returning (e.g., call the model's .dict()
or .dict(by_alias=True) as appropriate, or serialize nested fields) so the
FastMCP tool returns JSON-serializable data instead of a Pydantic object.
2 participants
Currently
workflow-runs-listtool only works for the in-memory state of the MCPApp, which won't surface previous runs or persist across restarts. This change makes it work by querying temporal directly.Summary by CodeRabbit
New Features
Bug Fixes
Refactor