Print text to console and record it as result output. Use for actual Data Agent results that should appear in output.md: conclusions, plans,

Make a generic API call. Args: action: API action name. version: API version. params: Request parameters

Get a field value trying multiple possible key names.

Get the status of a session. Args: session_id: The session ID. agent_id: The agent ID (optional, but used to construc

Get a field value trying multiple possible key names (case-insensitive). Args: obj: The dictionary to search *names: Possible fie

Create configuration from environment variables. Args: dotenv_path: Optional path to .env file to load. Returns:

Resolve the DMSUnit for the current tenant. Resolution priority: 1. Explicit config/environment override (``DATA_AGENT_DMS_UNIT``).

Run a synchronous function in an executor. Args: func: The function to run. *args: Positional arguments.

Print text to console only (progress.log) without recording to output.md. Use for diagnostic / process information: user query echo, plan step

Get a field value trying multiple possible key names.

Update the last used timestamp.

Check if session is in an active/usable state (RUNNING, IDLE or WAIT_INPUT).

Stream response tokens to stdout in real-time. Uses StreamState to track cross-event context (plan progress, output_conclusion accumulation,

Create a new session or reuse an existing one. Args: session_id: Optional existing session ID to reuse. agent_id: Opt

Send a message to the Data Agent. Args: agent_id: The agent ID. session_id: The session ID. message: The

Stream chat content from Data Agent using SSE. Automatically reconnects from the last checkpoint on SSL/connection errors, enabling r

Write to both plain text and JSONL logs. Args: text: Text to write to the plain text log. data: Optional structured d

Return a copy of ``value`` with credentials redacted for debug logs.

Close open log file handles.

Close the structured logging files.

Return and clear the pending step label for use as conclusion header.

Wrap analysis conclusion text with visual borders. If output_dir is provided, extracts inline base64 images from the text, saves them to outp

Return whether verbose API debug logging is enabled.

Resolve workspace ID with priority: explicit > env > cache > API call. Args: explicit: Explicitly provided workspace ID (highest

Initialize structured logging to progress.log and progress.jsonl. Args: output_dir: Directory where progress files should be created

List files associated with a session. Args: session_id: The session ID. file_category: Optional filter, e.g. "WebRepo

Build DataSource from command-line arguments.

Attach to an existing session's SSE stream without sending a message. Streams all incoming events to stdout in real-time using the same forma

Extract all JSON objects/arrays from a mixed content string. Handles cases like: - Plain text before JSON: "some text {...}" - Multiple J

Extract a list from an API response regardless of nesting style. Handles ``{Data: [...]}`` and ``{Data: {List/DataList/Content: [...]}}``. Al

Parse accumulated tool_call_response and format output. jupyter_cell results only shown in detail mode.

Format ask_report_render confirmation prompt.

Format a jupyter_cell data event into human-readable text. Decision logic: - Parse nb_file_outputs for real content - Skip cells whose on

Normalize blank optional strings to None.

Prepare request parameters with PascalCase keys. Args: params: Original request parameters api_action: API action nam

Generic worker execution with common error handling and session management. Args: args: Command line arguments data_source: Optio

Setup and run async worker process. Args: args: Command line arguments session: Session object containing session_id and agent_id

Convert to API request format.

Call dms-enterprise.ListTagMetaAsset via a short-lived OpenApiClient. Args: tag_name: The tag name to query assets for.

Execute batch preset queries.

Execute a single query with streaming output.

Convert SSE event to ContentBlock. Args: event: SSE event to convert. Returns: ContentBlock or None if not a

Convert SSE event to ContentBlock.

Flush any pending accumulators when the stream ends.

Format a list of insight objects (final conclusion).

Format a parsed JSON value (dict or list) for display.

Format a unix timestamp (seconds) to readable date string.

Format ask_plan confirmation prompt.

Handle data/plan events — show step progress. detail mode: full "[Plan] Step n/N: name" with description. summary mode: compact progress dot

Parse an SSE event. Args: event_type: The event type. data_str: The data JSON string. Returns: P

Parse an SSE event.

Route a single SSE event to the appropriate output handler. Args: event: The SSE event to handle. output_mode: "summary" (minimal

Print list of files generated by the Agent.

Calculate SHA256 hash and return hex string.

Wrapper for _stream_response that handles data_source parameter properly.

Recursively convert all dictionary keys to PascalCase. Args: obj: Input object (dict, list, or primitive) exclude_paths: List of

Create a new Data Agent session. Args: database_id: Optional database ID to bind to the session. title: Session title

Download a file from a URL and save locally. Args: download_url: Direct download URL. save_path: Local path to save t

Notify the service that file upload is complete. Args: file_id: The file ID from upload signature (uploadDir). filena

Get OSS upload signature for file upload. Args: filename: Name of the file to upload. file_size: Size of the file in

Get complete response by collecting all SSE events. Collects all `data` events (excluding tool_call_choices and metric_agent_config),

Handle worker process completion by updating status files. Args: session_dir: Session directory path need_confirm: Whether user c

Check if a file type is supported. Args: file_path: Path to the file. Returns: True if file type is supporte

Check if this is a worker process.

List databases in workspace via ListTagMetaAsset. Falls back to list_file_databases in API_KEY mode. Args: workspace_id:

List files associated with a session. Args: session_id: The session ID. file_category: Optional filter, e.g. "WebRepo

Prepare request body with PascalCase keys. Args: body: Original request body Returns: Body with PascalCase k

Process API response to convert keys to camelCase. Args: response: Original API response api_action: API action name

Generate signed headers for Alibaba Cloud API V3. Args: access_key_id: Access Key ID. access_key_secret: Access Key S

Upload a file for analysis. Args: file_path: Path to the local file. timeout: Upload timeout in seconds. Ret

Wait for session to reach RUNNING state. Args: session_id: The session ID. agent_id: The agent ID (optional).

Write structured data to JSONL file. Args: data: Dictionary containing structured log data

Write text to progress log file. Args: text: Text to write to the plain text log

Single-connection async streaming attempt.

Async stream with AK/SK authentication via default credential chain.

Async stream with API_KEY authentication.

Build an OpenApiClient for dms-enterprise using the default credential chain.

Invoke GetActiveRouteUnit with the given params. Returns the response body dict.

Make an API call using AK/SK authentication.

Make an API call using API_KEY authentication.

Call dms-enterprise.GetActiveRouteUnit via a short-lived OpenApiClient. The dms-enterprise endpoint differs from the Data Agent endpoint, so

Call InitDataAgentPersonalWorkspace to get the user's personal workspace ID. Returns: The workspace ID string. Raises:

Describe a custom agent in detail.

List custom agents.

Confirm the high-risk import operation before calling the API.

Main dispatcher: routes by event_type to specialised handlers.

Single-connection streaming attempt.

Stream with AK/SK authentication via default credential chain.

Stream with API_KEY authentication.

Extract inline base64 images from text and save to disk. Returns cleaned text with images replaced by file-path references.

Search for Route.RegionId in the response body. Per OpenAPI spec, the Route object (backend: "data") may contain {Uid, RegionId, Status}. Onl

Format a plan data event showing step progress. Expected plan_data structure (from API): {"current_step": 2, "plan_status": "...", "plans":

Format recommended follow-up questions. Expected JSON: {"questions": ["q1", "q2", ...]} or a list of strings.

Format a status_change event's JSON content. Expected JSON: {"previous": "PLANNING", "current": "STEP_EXECUTION", "current_task": "..."}

Format table summary list for display.