DAS Agent Chat Send natural language questions to the Alibaba Cloud DAS (Database Autonomy Service) Agent and receive diagnostic results. Pricing and Free Tier This is a paid service with a free tier for trial usage. - Free Tier : When is not set, the script omits the parameter and the API will use a default Agent ID which comes with a limited free quota for trial purposes. - Paid Usage : For production workloads or higher usage volumes, purchase a DAS Agent subscription and set your own to bind your dedicated agent and quota. Recommendation : Start with the free tier (default Agent ID) to ev…

, session_id):\n raise ValueError(\n \"Session ID contains invalid characters. \"\n \"Only alphanumeric characters and hyphens are allowed.\"\n )\n \n return session_id\n\n def chat(self, message: str, session_id: str = None) -> None:\n \"\"\"Send a message and receive streaming response.\"\"\"\n # Validate inputs\n message = self._validate_message(message)\n session_id = self._validate_session_id(session_id)\n\n # Reset state\n self.message_roles.clear()\n self.accumulated_text = \"\"\n self.accumulated_tool_result = \"\"\n self.current_message_id = None\n self.current_tool_id = None\n\n # Use provided session_id, or generate a new one\n if session_id is None:\n session_id = str(uuid.uuid4())\n\n request = SignatureRequest(\"POST\", \"/\", self.host, self.action, self.version)\n\n # Build Message JSON (using compact format, no spaces)\n message_json = {\n \"id\": str(uuid.uuid4()),\n \"role\": \"user\",\n \"content\": [\n {\n \"type\": \"text\",\n \"text\": message\n }\n ]\n }\n\n # Build form data (reference: curl invocation)\n form_data = OrderedDict()\n form_data[\"Format\"] = \"JSON\"\n form_data[\"SecureTransport\"] = \"true\"\n # Use compact JSON format, no spaces\n form_data[\"Message\"] = json.dumps(message_json, ensure_ascii=False, separators=(',', ':'))\n form_data[\"SourceTlsVersion\"] = \"TLSv1.2\"\n form_data[\"AcceptLanguage\"] = \"zh-CN\"\n # Only include AgentId if provided; omitting it uses the default Agent ID with free quota\n if self.agent_id:\n form_data[\"AgentId\"] = self.agent_id\n form_data[\"SessionId\"] = session_id\n\n # Manually build URL-encoded string to ensure correct encoding\n body_parts = []\n for key, value in form_data.items():\n # URL-encode the value using quote's safe parameter\n encoded_value = quote(str(value), safe='')\n body_parts.append(f\"{key}={encoded_value}\")\n request.body = \"&\".join(body_parts).encode('utf-8')\n request.headers[\"content-type\"] = \"application/x-www-form-urlencoded\"\n\n self._get_authorization(request)\n\n # Output session ID only after credential resolution succeeds.\n if self.mode == self.MODE_JSON:\n self._emit_json({\"type\": \"session\", \"session_id\": session_id})\n elif self.mode == self.MODE_PIPE:\n self._emit(f\"SESSION: {session_id}\")\n else:\n self._emit(f\"[Session: {session_id}]\")\n\n self._call_api(request)\n\n def _call_api(self, request: SignatureRequest) -> None:\n \"\"\"Call the API and handle streaming response.\"\"\"\n url = f\"https://{request.host}{request.canonical_uri}\"\n if request.query_param:\n url += \"?\" + urlencode(request.query_param, doseq=True, safe=\"*\")\n\n headers = dict(request.headers)\n data = request.body\n\n try:\n response = requests.request(\n method=request.http_method,\n url=url,\n headers=headers,\n data=data,\n stream=True,\n timeout=300\n )\n\n if response.status_code != 200:\n error_msg = f\"HTTP {response.status_code}: {response.text[:500]}\"\n self._progress(f\"HTTP error: {response.status_code}\")\n self._progress(f\"Response content: {response.text}\")\n if self.mode == self.MODE_JSON:\n self._emit_json({\"type\": \"error\", \"code\": response.status_code, \"message\": error_msg})\n return\n\n for line in response.iter_lines(decode_unicode=True):\n if line:\n self._process_sse_line(line)\n\n except requests.exceptions.Timeout:\n self._progress(\"Request timed out\")\n if self.mode == self.MODE_JSON:\n self._emit_json({\"type\": \"error\", \"code\": \"TIMEOUT\", \"message\": \"Request timed out\"})\n except requests.exceptions.ConnectionError:\n self._progress(\"Connection failed\")\n if self.mode == self.MODE_JSON:\n self._emit_json({\"type\": \"error\", \"code\": \"CONNECTION_ERROR\", \"message\": \"Connection failed\"})\n except requests.exceptions.HTTPError as e:\n self._progress(f\"HTTP error: {e}\")\n if hasattr(e.response, 'text'):\n self._progress(f\"Error details: {e.response.text}\")\n if self.mode == self.MODE_JSON:\n self._emit_json({\"type\": \"error\", \"code\": \"HTTP_ERROR\", \"message\": str(e)})\n except Exception as e:\n self._progress(f\"Unknown error: {e}\")\n import traceback\n traceback.print_exc(file=sys.stderr)\n if self.mode == self.MODE_JSON:\n self._emit_json({\"type\": \"error\", \"code\": \"UNKNOWN\", \"message\": str(e)})\n\n def _process_sse_line(self, line: str) -> None:\n \"\"\"Process a single SSE data line.\"\"\"\n if not line.startswith('data:'):\n return\n\n data_content = line[5:]\n if data_content == '[DONE]':\n return\n\n try:\n json_data = json.loads(data_content)\n except json.JSONDecodeError:\n self._progress(f\"[JSON parse error] Raw data: {data_content[:100]}...\")\n if self.mode == self.MODE_JSON:\n self._emit_json({\"type\": \"error\", \"code\": \"JSON_PARSE\", \"message\": f\"Failed to parse SSE data: {data_content[:200]}\"})\n return\n\n event_type = json_data.get('Type')\n handler = {\n 'TEXT_MESSAGE_START': self._on_text_message_start,\n 'TEXT_MESSAGE_CONTENT': self._on_text_message_content,\n 'TEXT_MESSAGE_END': self._on_text_message_end,\n 'TOOL_CALL_START': self._on_tool_call_start,\n 'TOOL_CALL_ARGS': self._on_tool_call_args,\n 'TOOL_CALL_RESULT': self._on_tool_call_result,\n 'TOOL_CALL_CHUNK': self._on_tool_call_chunk,\n 'TOOL_CALL_END': self._on_tool_call_end,\n 'ACTIVITY_DELTA': self._on_activity_delta,\n 'CUSTOM': self._on_custom,\n 'RUN_STARTED': lambda d: None,\n 'RUN_FINISHED': lambda d: None,\n }.get(event_type)\n\n if handler:\n handler(json_data)\n elif 'Answer' in json_data:\n # Legacy format compatibility\n self._on_legacy_answer(json_data)\n\n # --- SSE Event Handlers ---\n\n def _on_text_message_start(self, data):\n message_id = data.get('MessageId')\n role = data.get('Role', '')\n if message_id and role:\n self.message_roles[message_id] = role\n if role == 'assistant':\n self.current_message_id = message_id\n self.accumulated_text = \"\"\n\n def _on_text_message_content(self, data):\n delta = data.get('Delta', '')\n message_id = data.get('MessageId')\n if not delta:\n return\n\n # Determine if this is an assistant message\n role = self.message_roles.get(message_id, '') if message_id else 'assistant'\n if role != 'assistant':\n return\n\n # Accumulate text (all modes need this)\n if message_id == self.current_message_id or not message_id:\n self.accumulated_text += delta\n\n # CLI mode: stream to stdout in real-time\n # PIPE mode: accumulate only; we print with delimiters at message end\n if self.mode == self.MODE_CLI:\n self._emit(delta, end='')\n\n def _on_text_message_end(self, data):\n message_id = data.get('MessageId')\n if not message_id or message_id not in self.message_roles:\n return\n\n role = self.message_roles[message_id]\n del self.message_roles[message_id]\n\n if role != 'assistant' or message_id != self.current_message_id:\n return\n\n text = self.accumulated_text.strip()\n self.current_message_id = None\n self.accumulated_text = \"\"\n\n if not text:\n return\n\n # Output the complete assistant message\n if self.mode == self.MODE_JSON:\n self._emit_json({\"type\": \"message\", \"role\": \"assistant\", \"content\": text})\n elif self.mode == self.MODE_PIPE:\n # PIPE mode: wrap the answer in clear delimiters on stdout so the host agent\n # can identify and relay the real DAS response without ambiguity.\n self._emit(\"\\n=== DAS AGENT RESPONSE ===\")\n self._emit(text)\n self._emit(\"=== END RESPONSE ===\")\n else:\n # CLI mode: already streamed, just add a newline separator\n self._emit(\"\\n\")\n\n def _on_tool_call_start(self, data):\n tool_id = data.get('ToolCallId')\n # Extract tool name from SSE event - the field is \"ToolCallName\" based on actual API response\n tool_name = data.get('ToolCallName') or data.get('Name') or data.get('tool_name') or 'unknown_tool'\n\n self.tool_call_data[tool_id] = {\n 'name': tool_name,\n 'args': '',\n 'result': '',\n }\n self.current_tool_id = tool_id\n self.last_tool_name = tool_name # Save for fallback in TOOL_CALL_RESULT\n self.accumulated_tool_result = \"\"\n\n if self.mode == self.MODE_JSON:\n self._emit_json({\"type\": \"tool_call\", \"tool\": tool_name, \"status\": \"started\"})\n elif self.mode == self.MODE_PIPE:\n self._emit(f\"[tool] {tool_name} started\")\n else:\n # CLI mode: show tool call with newline\n self._emit(f\"\\nCalling tool [{tool_name}]...\")\n\n def _on_tool_call_args(self, data):\n tool_id = data.get('ToolCallId')\n delta = data.get('Delta', '')\n if tool_id in self.tool_call_data:\n self.tool_call_data[tool_id]['args'] += delta\n\n def _on_tool_call_result(self, data):\n tool_id = data.get('ToolCallId')\n delta = data.get('Delta', '')\n direct_result = data.get('Result')\n content = data.get('Content', '')\n\n # Determine result content and source\n result_content = ''\n result_source = ''\n if delta:\n result_content = delta\n result_source = 'Delta'\n elif direct_result is not None:\n result_content = json.dumps(direct_result, ensure_ascii=False)\n result_source = 'Result'\n elif content:\n result_content = content\n result_source = 'Content'\n\n if not result_content:\n return\n\n # Accumulate into tool data\n if tool_id and tool_id in self.tool_call_data:\n self.tool_call_data[tool_id]['result'] += result_content\n\n # Accumulate for current tool\n if tool_id == self.current_tool_id:\n self.accumulated_tool_result += result_content\n\n # Get tool name for output\n # Note: SSE may send TOOL_CALL_RESULT after TOOL_CALL_END, so tool_call_data may be deleted\n # Use last_tool_name as fallback\n tool_name = 'unknown_tool'\n if tool_id and tool_id in self.tool_call_data:\n tool_name = self.tool_call_data[tool_id]['name']\n elif self.current_tool_id and self.current_tool_id in self.tool_call_data:\n tool_name = self.tool_call_data[self.current_tool_id]['name']\n elif self.last_tool_name:\n # Fallback to last known tool name (handles RESULT after END scenario)\n tool_name = self.last_tool_name\n\n # Output result content (Content-type results contain actual API responses)\n if result_source == 'Content':\n if self.mode == self.MODE_CLI:\n self._emit(f\"\\n[Result]\")\n preview = result_content[:500] + \"...\" if len(result_content) > 500 else result_content\n self._emit(preview)\n elif self.mode == self.MODE_PIPE:\n # PIPE mode: print tool output inline on stdout\n preview = result_content[:300] + \"...\" if len(result_content) > 300 else result_content\n self._emit(f\"[tool_output] {preview}\")\n elif self.mode == self.MODE_JSON:\n # Emit tool output immediately for JSON mode\n if len(result_content) > 5000:\n self._emit_json({\"type\": \"tool_output\", \"tool\": tool_name, \"content\": result_content[:5000] + \"...(truncated)\"})\n else:\n self._emit_json({\"type\": \"tool_output\", \"tool\": tool_name, \"content\": result_content})\n\n def _on_tool_call_chunk(self, data):\n tool_id = data.get('ToolCallId')\n chunk = data.get('Chunk', '')\n if tool_id in self.tool_call_data:\n self.tool_call_data[tool_id]['result'] += str(chunk)\n\n def _on_tool_call_end(self, data):\n tool_id = data.get('ToolCallId')\n direct_result = data.get('Result')\n\n # Get tool info before deletion\n tool_name = 'unknown_tool'\n tool_args = ''\n if tool_id and tool_id in self.tool_call_data:\n tool_name = self.tool_call_data[tool_id]['name']\n tool_args = self.tool_call_data[tool_id]['args']\n # Also check if result was accumulated in tool_call_data\n if not self.accumulated_tool_result:\n self.accumulated_tool_result = self.tool_call_data[tool_id].get('result', '')\n del self.tool_call_data[tool_id]\n\n # Determine final result text\n result_text = \"\"\n if direct_result is not None:\n result_text = json.dumps(direct_result, ensure_ascii=False)\n elif self.accumulated_tool_result:\n result_text = self.accumulated_tool_result\n\n # Output tool result in JSON mode (always emit, even if result is empty)\n if self.mode == self.MODE_JSON:\n event = {\"type\": \"tool_result\", \"tool\": tool_name}\n if tool_args:\n event[\"args\"] = tool_args\n if result_text:\n # Truncate very long results to avoid overwhelming output\n if len(result_text) > 5000:\n event[\"content\"] = result_text[:5000] + \"...(truncated)\"\n else:\n event[\"content\"] = result_text\n self._emit_json(event)\n\n # Reset\n if tool_id == self.current_tool_id:\n self.current_tool_id = None\n self.accumulated_tool_result = \"\"\n\n # Progress indicator\n if self.mode == self.MODE_CLI:\n print(\" done\", flush=True)\n elif self.mode == self.MODE_PIPE:\n self._emit(f\"[tool] {tool_name} done\")\n\n def _on_activity_delta(self, data):\n # Show progress dots in CLI/PIPE modes; skip in JSON mode\n if self.mode in (self.MODE_CLI, self.MODE_PIPE):\n print(\".\", end=\"\", flush=True)\n\n def _on_custom(self, data):\n event_name = data.get('Name', '')\n value = data.get('Value', {})\n if event_name == 'error' and isinstance(value, dict):\n error_code = value.get('Code', 'unknown')\n error_msg = value.get('Message', 'Unknown error')\n self._progress(f\"\\n[Error {error_code}] {error_msg}\")\n if self.mode == self.MODE_JSON:\n self._emit_json({\"type\": \"error\", \"code\": error_code, \"message\": error_msg})\n\n def _on_legacy_answer(self, data):\n answer = data.get('Answer', '')\n if not answer:\n return\n if self.mode == self.MODE_JSON:\n self._emit_json({\"type\": \"message\", \"role\": \"assistant\", \"content\": answer})\n else:\n self._emit(answer, end='')\n\n\ndef main():\n parser = argparse.ArgumentParser(description=\"Call Alibaba Cloud DAS Agent Chat API\")\n parser.add_argument(\"--question\", required=True, help=\"The question to send to the Agent\")\n parser.add_argument(\"--session\", help=\"Session ID for maintaining conversation context\")\n parser.add_argument(\"--json\", action=\"store_true\", help=\"JSONL output: one JSON object per line, machine-readable\")\n parser.add_argument(\"--pipe\", action=\"store_true\",\n help=\"Agent-friendly mode: progress/tool noise to stderr, answer delimited on stdout\")\n args = parser.parse_args()\n\n if args.json:\n mode = DasAgentChatClient.MODE_JSON\n elif args.pipe:\n mode = DasAgentChatClient.MODE_PIPE\n else:\n mode = DasAgentChatClient.MODE_CLI\n\n try:\n client = DasAgentChatClient(mode=mode)\n client.chat(args.question, session_id=args.session)\n except ValueError as e:\n print(f\"Configuration error: {e}\", file=sys.stderr)\n sys.exit(1)\n except ImportError as e:\n print(\n \"Dependency error: install project dependencies first (missing Alibaba Cloud Credentials SDK).\",\n file=sys.stderr,\n )\n print(f\"Details: {e}\", file=sys.stderr)\n sys.exit(1)\n except Exception as e:\n print(f\"Runtime error: {e}\", file=sys.stderr)\n sys.exit(1)\n\n\nif __name__ == \"__main__\":\n main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":31043,"content_sha256":"d9ad92fbcd951043410f8f7a18887f26248d9685ee41178bea4e8fd4d7ad56c1"},{"filename":"scripts/pyproject.toml","content":"[project]\nname = \"das-agent-skill\"\nversion = \"0.1.0\"\ndescription = \"Client for calling Alibaba Cloud DAS Agent Chat API\"\nrequires-python = \">=3.10\"\ndependencies = [\n \"alibabacloud_credentials==1.0.2\",\n \"pytz==2025.2\",\n \"requests==2.32.5\",\n]\n\n[project.optional-dependencies]\ntest = [\n \"pytest==7.0.0\",\n \"pytest-cov==4.0.0\",\n]\n","content_type":"text/plain; charset=utf-8","language":"toml","size":340,"content_sha256":"a30dc1b369067733db29c41ecaed3c1cbde1a514ce34dc292808161fec87a5c8"}],"content_json":{"type":"doc","content":[{"type":"heading","attrs":{"level":1},"content":[{"text":"DAS Agent Chat","type":"text"}]},{"type":"paragraph","content":[{"text":"Send natural language questions to the Alibaba Cloud DAS (Database Autonomy Service) Agent and receive diagnostic results.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Pricing and Free Tier","type":"text"}]},{"type":"paragraph","content":[{"text":"This is a paid service with a free tier for trial usage.","type":"text","marks":[{"type":"strong"}]}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Free Tier","type":"text","marks":[{"type":"strong"}]},{"text":": When ","type":"text"},{"text":"ALIBABA_CLOUD_DAS_AGENT_ID","type":"text","marks":[{"type":"code_inline"}]},{"text":" is not set, the script omits the ","type":"text"},{"text":"AgentId","type":"text","marks":[{"type":"code_inline"}]},{"text":" parameter and the API will use a ","type":"text"},{"text":"default Agent ID","type":"text","marks":[{"type":"strong"}]},{"text":" which comes with a limited free quota for trial purposes.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Paid Usage","type":"text","marks":[{"type":"strong"}]},{"text":": For production workloads or higher usage volumes, purchase a DAS Agent subscription and set your own ","type":"text"},{"text":"ALIBABA_CLOUD_DAS_AGENT_ID","type":"text","marks":[{"type":"code_inline"}]},{"text":" to bind your dedicated agent and quota.","type":"text"}]}]}]},{"type":"blockquote","content":[{"type":"paragraph","content":[{"text":"Recommendation","type":"text","marks":[{"type":"strong"}]},{"text":": Start with the free tier (default Agent ID) to evaluate the service. Once you decide to adopt it for production, purchase a subscription and configure your own Agent ID.","type":"text"}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Environment Variables","type":"text"}]},{"type":"paragraph","content":[{"text":"The script requires Alibaba Cloud credentials resolvable via the ","type":"text"},{"text":"default credential chain","type":"text","marks":[{"type":"link","attrs":{"href":"https://www.alibabacloud.com/help/en/sdk/developer-reference/v2-manage-python-access-credentials","title":null}}]},{"text":". The DAS Agent ID is ","type":"text"},{"text":"optional","type":"text","marks":[{"type":"strong"}]},{"text":" — if not provided, the ","type":"text"},{"text":"AgentId","type":"text","marks":[{"type":"code_inline"}]},{"text":" parameter will be omitted and the API will use a default Agent ID with limited free quota.","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"# Optional: Set your own Agent ID after purchasing DAS Agent service\nexport ALIBABA_CLOUD_DAS_AGENT_ID=\"\u003cagent_id>\" # Obtain from DAS console (optional)","type":"text"}]},{"type":"paragraph","content":[{"text":"The Alibaba Cloud Credentials SDK automatically resolves credentials from multiple sources (environment variables, configuration files, ECS RAM roles, etc.). Refer to the ","type":"text"},{"text":"official credential configuration documentation","type":"text","marks":[{"type":"link","attrs":{"href":"https://www.alibabacloud.com/help/en/sdk/developer-reference/v2-manage-python-access-credentials","title":null}}]},{"text":" for setup instructions.","type":"text"}]},{"type":"paragraph","content":[{"text":"If you have purchased a DAS Agent subscription, create and manage your Agent ID at: https://das.console.aliyun.com/","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Troubleshooting","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Credential Resolution Failed","type":"text"}]},{"type":"paragraph","content":[{"text":"If the script exits with a credential-related error, it means the Alibaba Cloud Credentials SDK could not resolve a usable credential from its default provider chain.","type":"text"}]},{"type":"paragraph","content":[{"text":"Supported credential sources:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Environment variables (see ","type":"text"},{"text":"official documentation","type":"text","marks":[{"type":"link","attrs":{"href":"https://www.alibabacloud.com/help/en/sdk/developer-reference/v2-manage-python-access-credentials","title":null}}]},{"text":")","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Local profile files: ","type":"text"},{"text":"~/.aliyun/config.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" or ","type":"text"},{"text":"~/.alibabacloud/credentials.ini","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"ECS RAM Role metadata when running on an Alibaba Cloud ECS instance","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Common cases:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Credential environment variables are empty or missing — configure them according to the official documentation.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Errors mentioning ","type":"text"},{"text":"~/.aliyun/config.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" or ","type":"text"},{"text":"~/.alibabacloud/credentials.ini","type":"text","marks":[{"type":"code_inline"}]},{"text":" The SDK attempted local profile-based credentials, but the files were missing or invalid. Create/fix the default profile if you want to use local profiles.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Errors mentioning ","type":"text"},{"text":"100.100.100.200","type":"text","marks":[{"type":"code_inline"}]},{"text":" The SDK attempted ECS metadata. This is expected on ECS, but usually a local-machine misconfiguration elsewhere.","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"For local development, if you are not using ECS RAM Role credentials, you can explicitly disable ECS metadata lookup:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"export ALIBABA_CLOUD_ECS_METADATA_DISABLED=true","type":"text"}]},{"type":"paragraph","content":[{"text":"This avoids confusing ","type":"text"},{"text":"100.100.100.200","type":"text","marks":[{"type":"code_inline"}]},{"text":" metadata connection errors on non-ECS machines and makes missing-credential failures easier to read.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Invocation","type":"text"}]},{"type":"paragraph","content":[{"text":"Run from the ","type":"text"},{"text":"scripts/","type":"text","marks":[{"type":"code_inline"}]},{"text":" directory of this skill:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"cd scripts\n\n# Pipe mode (RECOMMENDED for agents) — clean output: progress to stderr, answer clearly delimited on stdout\nuv run call_das_agent.py --question \"\u003cuser's question>\" --pipe\n\n# Default mode (CLI chat UI) — real-time streaming with tool details\nuv run call_das_agent.py --question \"\u003cuser's question>\"\n\n# JSON mode — machine-readable JSONL, one JSON object per line on stdout\nuv run call_das_agent.py --question \"\u003cuser's question>\" --json\n\n# Multi-turn conversation — reuse the server-assigned session ID to maintain context\nuv run call_das_agent.py --question \"List my instances\" --pipe # Returns session_id on first line\n# Extract session_id (line starting with \"SESSION:\"), then reuse it:\nuv run call_das_agent.py --question \"Check the first one\" --session \"\u003csession_id_from_above>\" --pipe","type":"text"}]},{"type":"paragraph","content":[{"text":"Always use ","type":"text","marks":[{"type":"strong"}]},{"text":"--pipe","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" when invoking as an agent.","type":"text","marks":[{"type":"strong"}]},{"text":" It routes all progress/tool-call noise to stderr and writes only the DAS answer to stdout, wrapped in clear delimiters — making the real response impossible to miss.","type":"text"}]},{"type":"paragraph","content":[{"text":"Prefer ","type":"text"},{"text":"--json","type":"text","marks":[{"type":"code_inline"}]},{"text":" when you need to parse the response programmatically. For JSON event types and output mode details, see ","type":"text"},{"text":"references/api-reference.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/api-reference.md","title":null}}]},{"text":".","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Behavioral Notes","type":"text"}]},{"type":"paragraph","content":[{"text":"DAS Agent internally orchestrates multiple API calls and tool invocations to answer a single question. This has important implications:","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Long-running tasks","type":"text","marks":[{"type":"strong"}]},{"text":": Complex diagnostics (multi-instance inspection, comprehensive health checks, batch SQL analysis) can take several minutes up to 30 minutes, because DAS Agent sequentially calls monitoring APIs, runs diagnostics, and synthesizes results. Inform the user before starting and provide periodic progress updates.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Instance enrollment","type":"text","marks":[{"type":"strong"}]},{"text":": The target database instance must be enrolled under the DAS Agent. If you see error code ","type":"text"},{"text":"-1810006","type":"text","marks":[{"type":"code_inline"}]},{"text":", it means the agent is not associated with any instance — guide the user to the ","type":"text"},{"text":"DAS console settings","type":"text","marks":[{"type":"link","attrs":{"href":"https://das.console.aliyun.com/?aes_debug=#/das-agent?currentView=settings","title":null}}]},{"text":" to associate instances.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Include instance ID in questions","type":"text","marks":[{"type":"strong"}]},{"text":": DAS Agent resolves instances by ID (e.g., ","type":"text"},{"text":"rm-bp1xxx","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"pc-2zeyyy","type":"text","marks":[{"type":"code_inline"}]},{"text":"). Always include the specific instance ID in the question for accurate results. If the user hasn't provided one, ask them or first query the instance list.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Parallel execution","type":"text","marks":[{"type":"strong"}]},{"text":": When diagnosing multiple instances, launch multiple script processes in parallel — each invocation is independent and stateless (unless sharing a session ID).","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Multi-turn conversations — always reuse the session ID when questions are related","type":"text","marks":[{"type":"strong"}]},{"text":": If the user's questions are sequential or contextually connected (follow-up diagnostics, drill-down analysis, referencing a previous result, comparing findings), you ","type":"text"},{"text":"must","type":"text","marks":[{"type":"strong"}]},{"text":" pass ","type":"text"},{"text":"--session \u003csession_id>","type":"text","marks":[{"type":"code_inline"}]},{"text":" on every subsequent call. Starting a new session mid-conversation forces the DAS Agent to re-run all prior context from scratch, wastes time, and produces lower-quality answers.","type":"text"}]},{"type":"paragraph","content":[{"text":"Decision rule","type":"text","marks":[{"type":"strong"}]},{"text":": Default to reusing the session ID. Only start a new session when the user explicitly switches to a completely unrelated topic or asks to \"start over\".","type":"text"}]},{"type":"paragraph","content":[{"text":"The session ID is ","type":"text"},{"text":"server-assigned","type":"text","marks":[{"type":"strong"}]},{"text":" and returned as the very first line of every ","type":"text"},{"text":"--pipe","type":"text","marks":[{"type":"code_inline"}]},{"text":" invocation:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":""},"content":[{"text":"SESSION: \u003cuuid>","type":"text"}]},{"type":"paragraph","content":[{"text":"Extract it immediately after each call and carry it forward. In ","type":"text"},{"text":"--json","type":"text","marks":[{"type":"code_inline"}]},{"text":" mode it appears as ","type":"text"},{"text":"{\"type\": \"session\", \"session_id\": \"...\"}","type":"text","marks":[{"type":"code_inline"}]},{"text":" on the first line.","type":"text"}]},{"type":"paragraph","content":[{"text":"Examples of when reuse is ","type":"text"},{"text":"mandatory","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"\"List my instances\" → \"Check CPU on the first one\" → \"Why is it high?\"","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"\"Run a health check on rm-bp1xxx\" → \"Show me the top slow queries\"","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"\"What locks are held?\" → \"Kill that session\"","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"The DAS Agent retains the full conversation history server-side, so follow-up questions can be short and natural — no need to repeat instance IDs or prior context.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Output","type":"text"}]},{"type":"paragraph","content":[{"text":"For output mode comparison and format details, see ","type":"text"},{"text":"references/api-reference.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/api-reference.md","title":null}}]},{"text":".","type":"text"}]},{"type":"paragraph","content":[{"text":"After running the script, relay the complete stdout to the user verbatim.","type":"text","marks":[{"type":"strong"}]},{"text":" Do not summarize, paraphrase, or omit any part of the script's stdout. The DAS Agent's actual diagnostic answer is embedded in the output — the user must see it in full.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"For detailed API signature and SSE event documentation, see ","type":"text"},{"text":"references/api-reference.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/api-reference.md","title":null}}]},{"text":".","type":"text"}]}]},"metadata":{"date":"2026-06-05","name":"alibabacloud-das-agent","author":"@skillopedia","source":{"stars":133,"repo_name":"alibabacloud-aiops-skills","origin_url":"https://github.com/aliyun/alibabacloud-aiops-skills/blob/HEAD/skills/database/hdm/alibabacloud-das-agent/SKILL.md","repo_owner":"aliyun","body_sha256":"9806e18c5cad311f85070872f0cc6282a32f7fafea8fe9a8626b4e8ed9c91189","cluster_key":"0d9c23e9dc9f55cceb85ee15e8eb86ebf783d911b7fc9e5fc63d62e1e31dd3a9","clean_bundle":{"format":"clean-skill-bundle-v1","source":"aliyun/alibabacloud-aiops-skills/skills/database/hdm/alibabacloud-das-agent/SKILL.md","attachments":[{"id":"1022bf27-6d0d-574d-bf6c-e19c74e9ac3a","key":"uploads/10433ee7-ad12-4ae0-b34e-97553e46c6c8/1022bf27-6d0d-574d-bf6c-e19c74e9ac3a/attachment.md","path":"references/api-reference.md","size":4069,"sha256":"2a6f3a3de846d20fd69aadc105a5eb44039ca180bbf8c68e296ed7fa205da626","contentType":"text/markdown; charset=utf-8"},{"id":"9ef4a6a7-1497-51b7-9332-ff3cd38d8d34","key":"uploads/10433ee7-ad12-4ae0-b34e-97553e46c6c8/9ef4a6a7-1497-51b7-9332-ff3cd38d8d34/attachment.md","path":"references/ram-policies.md","size":1420,"sha256":"feae80eb84850d6986956cba863e70d937dc706d27319c72e78abcaee3157917","contentType":"text/markdown; charset=utf-8"},{"id":"2d4ed82d-7463-57fd-b475-64a17a3ef400","key":"uploads/10433ee7-ad12-4ae0-b34e-97553e46c6c8/2d4ed82d-7463-57fd-b475-64a17a3ef400/attachment.py","path":"scripts/call_das_agent.py","size":31043,"sha256":"d9ad92fbcd951043410f8f7a18887f26248d9685ee41178bea4e8fd4d7ad56c1","contentType":"text/x-python; charset=utf-8"},{"id":"87160542-5bfc-5d2e-bda6-4eefaba8f233","key":"uploads/10433ee7-ad12-4ae0-b34e-97553e46c6c8/87160542-5bfc-5d2e-bda6-4eefaba8f233/attachment.toml","path":"scripts/pyproject.toml","size":340,"sha256":"a30dc1b369067733db29c41ecaed3c1cbde1a514ce34dc292808161fec87a5c8","contentType":"text/plain; charset=utf-8"}],"bundle_sha256":"c9a12e28a6acc9c96c6bd216d551c39f6a8db3bbdceec69b231ac2bd6a4fd6be","attachment_count":4,"text_attachments":4,"attachment_storage":"skillopedia-attachments-v1","binary_attachments":0,"excluded_attachments":[]},"cluster_size":1,"skill_md_path":"skills/database/hdm/alibabacloud-das-agent/SKILL.md","import_metadata":{"date":"2026-06-05","author":"@skillopedia","version":"v1","category":"security","category_label":"Security"},"exact_dupes_collapsed_into_this":0},"license":"Apache-2.0","version":"v1","category":"security","metadata":{"async":true,"timeout":1800,"required_permissions":["das:Chat"]},"import_tag":"clean-skills-v1","description":"Diagnose and manage Alibaba Cloud databases through natural language. Use when users need to troubleshoot database performance issues (high CPU, slow queries, abnormal connections, lock waits), check instance status, analyze disk space, optimize SQL, run health inspections, or detect security baseline violations. Supports RDS (MySQL/PostgreSQL/SQL Server), PolarDB, MongoDB, Redis (Tair), and Lindorm. Trigger this skill even for casual descriptions like \"my database is slow\", \"can't connect to the database\", \"help me check this SQL\", or \"database disk is almost full\". Also suitable for consulting Alibaba Cloud-specific database features (e.g., PolarDB Serverless, DAS autonomy capabilities) and comparing product differences (RDS vs PolarDB). Do NOT use this skill for general SQL tutorials, non-Alibaba Cloud databases, or local database administration.\n","compatibility":"Requires uv (Python package manager) and HTTPS access to das.cn-shanghai.aliyuncs.com. Requires Alibaba Cloud credentials to be available through the default credential chain (AliyunHDMFullAccess permission). DAS Agent ID is optional.\n"}},"renderedAt":1782981525608}

DAS Agent Chat Send natural language questions to the Alibaba Cloud DAS (Database Autonomy Service) Agent and receive diagnostic results. Pricing and Free Tier This is a paid service with a free tier for trial usage. - Free Tier : When is not set, the script omits the parameter and the API will use a default Agent ID which comes with a limited free quota for trial purposes. - Paid Usage : For production workloads or higher usage volumes, purchase a DAS Agent subscription and set your own to bind your dedicated agent and quota. Recommendation : Start with the free tier (default Agent ID) to ev…