
Volcengine Cli
- 32 installs
- 16 repo stars
- Updated August 3, 2026
- volcengine/volcengine-skills
Helps with ai & agent building tasks.
About
volcengine-cli is a Claude Code skill for ai & agent building. It helps solo builders move faster with AI-assisted development.
- volcengine-cli
- AI & Agent Building
- AI-coding skill
Volcengine Cli by the numbers
- 32 all-time installs (skills.sh)
- +6 installs in the week ending Jul 20, 2026 (Skillselion tracking)
- Ranked #9,000 of 16,556 AI & Agent Building skills by installs in the Skillselion catalog
- Data as of Aug 4, 2026 (Skillselion catalog sync)
npx skills add https://github.com/volcengine/volcengine-skills --skill volcengine-cliAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 32 |
|---|---|
| repo stars | ★ 16 |
| Last updated | August 3, 2026 |
| Repository | volcengine/volcengine-skills ↗ |
What it does
Helps with ai & agent building tasks.
Files
Volcengine CLI Skill
Create and manage Volcengine cloud resources by calling Volcengine OpenAPIs through the ve command.
---
0. Install the ve CLI
If the ve command is not available on the system:
Option 1: npm (recommended)
npm i -g @volcengine/cliOption 2: GitHub Releases Download: https://github.com/volcengine/volcengine-cli/releases
Verify the installation: ve --version
---
1. Initialization (run at the start of every session)
Run the identity verification command to confirm that credentials are usable:
ve sts GetCallerIdentitySuccess — inform the user of the current account identity and region, then proceed with the task.
Switching regions later: once a profile is set up,--regionandVOLCENGINE_REGIONdo not override the region baked into it. Switch viave configure profile --profile <name>(useve configure listto see profiles). This does not apply to the--regionflag onve loginitself, which is required (see below).
Failure — no usable profile. Default plan: use ve login (Console Login, OAuth 2.0 + PKCE). Announce this to the user up front, and tell them they can say "use AK/SK", "use STS token", or "use SSO" to switch.
First check the ve version:
ve --versionDefault: Console Login (requires ve >= 1.0.42)
Use ve login --remote via the helper script scripts/ve_login_remote.sh. It handles the OAuth device-flow subprocess lifecycle (FIFO-bound stdin, URL extraction, code feeding, cleanup) so the agent doesn't have to.
Resolve the login region:
1. If the user named a region in this conversation, use it. 2. Else if VOLCENGINE_REGION is set, use it. 3. Else default to cn-beijing.
Default procedure — commit to this path; do NOT present a menu of login methods:
1. Announce + give the user an off-ramp, then immediately start:
"I'll start ve login --remote --region <region> now. Say 'use AK/SK' anytime to switch."2. Start the login subprocess and get the URL:
scripts/ve_login_remote.sh start <region>Prints a https://signin.volcengine.com/... URL on stdout. Forward it verbatim to the user with: "Open this URL in any browser, complete login, then send me the 'Authorization code' shown on the page."
3. When the user replies with the code, complete the flow:
scripts/ve_login_remote.sh complete <code>The script writes the code into the FIFO bound to the still-running ve, waits for ve to exit, then runs ve sts GetCallerIdentity to verify.
4. If the user interrupts (says "use AK/SK", "cancel", "this is taking too long", etc.):
scripts/ve_login_remote.sh abortThen switch to the chosen alternative below.
Critical rules — do NOT improvise OAuth:
- ❌ Do NOT present a menu like "A. URL B. local browser C. AK/SK". Commit to
--remote; let the user interrupt to switch. - ❌ Do NOT pre-fetch the URL by running
ve login --remoteand exiting. The PKCE challenge dies with the subprocess; the URL becomes useless and the next attempt fails with PKCE mismatch. - ❌ Do NOT construct
signin.volcengine.com/authorize/...URLs yourself. - ❌ Do NOT decode, parse, or transform the user's reply. Whatever they paste IS the authorization code — pass it verbatim to
complete <code>. No base64 decode (even if the string looks base64-shaped), no URL query-string parsing, no extractingcode=from callback URLs. - ❌ Do NOT pipe codes in (
echo "<code>" | ve login --remote). The code arrives before the user completes browser login and fails verification. - ❌ Do NOT spawn parallel
ve loginsubprocesses. One at a time, tracked by the helper. - ❌ Do NOT call
ve login --remotedirectly. Always go throughscripts/ve_login_remote.sh. The script binds ve's stdin to a FIFO so the still-running ve can receive the user's code via a separatecompletecall. Calling ve directly orphans the subprocess and breaks the code-feeding step. - ✅ Use
scripts/ve_login_remote.sh start/complete/abort.
Switching region mid-flow:scripts/ve_login_remote.sh abort, thenstart <new-region>.
No browser on any device (true offline / CI): skip ve login, fall back to AK/SK below.If ve login fails (network error, non-interactive terminal, version too old), or the user explicitly asks for a different method, fall back to one of the alternatives below. If installed via npm, upgrade with npm i -g @volcengine/cli@latest.
Alternative: AK/SK (long-term credentials, for CI/CD or scripting)
Ask the user for AccessKey and SecretKey, then:
ve configure set --profile default --region cn-beijing \
--endpoint open.volcengineapi.com \
--access-key <AK> --secret-key <SK>For STS (temporary) credentials, also pass --session-token <TOKEN>.
Alternative for the current shell only: export VOLCENGINE_ACCESS_KEY, VOLCENGINE_SECRET_KEY, VOLCENGINE_REGION, optionally VOLCENGINE_SESSION_TOKEN.
Alternative: SSO / Cloud Identity Center (requires ve >= 1.0.38, for enterprise federation)
Three-step setup; ask the user for the SSO start URL and session name first:
ve configure sso-session --name <session-name> \
--start-url https://<sso-host>/userportal \
--region cn-beijing \
--registration-scopes cloudidentity:account:access,offline_access
ve configure sso --profile <profile-name> --sso-session <session-name>
ve configure profile --profile <profile-name>Then ve sso login --sso-session <session-name> (use --no-browser on headless machines).
Credential safety
- Never read `~/.volcengine/config.json` — it contains AK/SK and session tokens.
- When running
ve configure setwith--secret-key, prefer letting the user paste and run the command in their own shell rather than executing it via Claude — secrets passed as command-line arguments end up in shell history and process listings. - Never echo AK/SK, secret keys, or session tokens back to the user in plain text.
---
2. Safety Rules (mandatory)
Read/Write Classification
| Level | Operation Types | Behavior |
|---|---|---|
| Read-only | Describe\ / List\ / Get\ / Query\ | Execute directly, no confirmation needed |
| Write | Create\ / Run\ / Allocate\ / Attach\ / Associate\ / Authorize\ | Show the full command and wait for user confirmation |
| Destructive | Delete\ / Terminate\ / Release\ / Revoke\ / Modify\ / Stop\ / Detach\* | Show command + impact summary; require user confirmation |
Core Principles
1. Default to read-only — unless the user explicitly requests a change, execute in read-only mode 2. DryRun first — if a write/destructive operation supports --DryRun true, run a DryRun to preview the plan, then confirm before executing 3. Confirm before executing — show the full command for write operations and wait for approval 4. Protect credentials — never read ~/.volcengine/config.json; never expose access-key, secret-key, or session-token in output
DryRun Notes
A successful DryRun validation returns exit code 1 (non-zero) with DryRunOperation in stderr. This is expected behavior:
output=$(ve <svc> <action> --DryRun true ... 2>&1)
if echo "$output" | grep -q "DryRunOperation"; then
echo "Parameter validation passed"
fi---
3. Locate APIs and Retrieve Parameters
Locate the API (find the service name + Action name)
Step 1: Service name + Action known? -> Use them directly; skip to "Retrieve parameters"
Step 2: Service name known, Action unknown?
-> ve <service> 2>&1 | grep -i <keyword>
Step 3: Service name also unknown?
-> ve 2>&1 | grep -i <service keyword>
Step 4: None of the above work?
-> python3 scripts/find_api.py <keyword>Retrieve parameters (once the Action is known)
Choose a strategy based on operation type:
| Operation Type | Strategy | Rationale |
|---|---|---|
| Read-only (Describe/List/Get) | ve <svc> <action> --help | Few, simple parameters — names alone are sufficient |
| Write/destructive (Create/Run/Delete, etc.) | scripts/fetch_swagger.py for full docs | Many parameters, nested structures — need required fields, examples, and descriptions |
| Still unclear after `--help` | Supplement with scripts/fetch_swagger.py | Use whenever parameter meaning is uncertain |
*Errors like `Invalid / Missing`* | Recheck with scripts/fetch_swagger.py | On InvalidParameter, InvalidXxx.NotFound, or MissingParameter, verify parameter names, required fields, and value ranges |
# Read-only — --help is sufficient
ve ecs DescribeInstances --help
# Write — retrieve full documentation
python3 scripts/fetch_swagger.py --service ecs --action RunInstancesve command name and API version relationship
- Default version -> ve command = base service name (e.g.,
iam) - Non-default version -> ve command =
service name + version without hyphens(e.g.,iamv2021-08-01 ->iam20210801) - When in doubt:
ve 2>&1 | grep <service>to confirm
Python helper usage
# Search for an API (when the service name is unknown)
python3 scripts/find_api.py <keyword> [--limit N]
# Get full API parameter documentation (when descriptions/examples are needed)
python3 scripts/fetch_swagger.py --service <ServiceCode> --action <ActionName>
# List all APIs for a service
python3 scripts/fetch_swagger.py --service <ServiceCode> --list
# Call Extension APIs that public API Explorer/ve does not expose
python3 scripts/call_extend_api.py --list
python3 scripts/call_extend_api.py --describe QueryMetricsAlways pass the base service name to scripts/fetch_swagger.py (e.g.,--service iam, notiam20210801) — the script auto-detects the version.
Extension APIs
Some extension APIs are not exposed through ve. For those, consult references/extend-apis.md and use:
python3 scripts/call_extend_api.py --api <APIName> --params '{"Key":"Value"}'The helper resolves service, version, method, endpoint, content type, and credentials from its registry/environment. Credentials prefer VOLCENGINE_ACCESS_KEY/VOLCENGINE_SECRET_KEY; when those environment variables are missing, it warns the user and falls back to the Python SDK's CLIConfigCredentialProvider so ve profiles from ak, ramrolearn, oidc, ecsrole, sso, and console-login modes can be reused. Apply the same read/write/destructive confirmation rules before running extension APIs.
---
4. Execute API Calls
Basic Format
ve <ServiceCode> <ActionName> --ParamName "value"Parameter Passing Rules
Determine the format from --help output:
- Flat parameter format:
--helplists individual--Key typeentries (e.g., ECS, VPC, IAM) -> pass with--Key "value" - Array parameters: prefer the numbered CLI form shown by
--help, such as--InstanceIds.1 "$instance_id"or--SubnetIds.1 "$subnet_id". Do not assume JSON-array strings are accepted by every action. - JSON format:
--helponly shows--body '{...}'(e.g., Redis, CR, and other POST APIs) -> pass with--body '{...}'
# Flat parameters — nested fields use dot notation; arrays use .N index (starting from 1)
ve ecs RunInstances --ZoneId "cn-beijing-a"
ve ecs RunInstances --NetworkInterfaces.1.SubnetId "subnet-xxxx"
ve ecs RunInstances --Tags.1.Key "publish-by" --Tags.1.Value "deploy-skill"
# JSON format (when --help only shows --body)
ve redis CreateDBInstance --body '{"InstanceName":"demo","RegionId":"cn-beijing","ConfigureNodes":[{"AZ":"cn-beijing-a"}],"ShardedCluster":0,"NodeNumber":2,"ShardCapacity":1024,"ShardNumber":1,"EngineVersion":"6.0","SubnetId":"subnet-xxxx","VpcId":"vpc-xxxx","Password":"<secret>","Tags":[{"Key":"publish-by","Value":"deploy-skill"}]}'Response Format
// Success
{ "ResponseMetadata": { "RequestId": "..." }, "Result": { ... } }
// Failure
{ "ResponseMetadata": { "Error": { "Code": "...", "Message": "..." } } }Async Resource Creation Requires Polling
Some resources (VKE clusters, RDS instances, ECS instances, etc.) take several minutes to create. After creation, poll the Describe endpoint until the resource reaches the desired status before proceeding.
Creating sub-resources (e.g., security groups) immediately after VPC creation may fail with InvalidVpc.InvalidStatus. Create sub-resources sequentially (subnet first, then security group), or wait a few seconds and retry.# General polling pattern: check every 30 seconds until the target status is reached
while true; do
cur_status=$(ve <svc> Describe<Resource> --<IdParam> "xxx" 2>&1 | grep -o '"Status":"[^"]*"')
echo "$(date +%H:%M:%S) $cur_status"
echo "$cur_status" | grep -q '"Status":"Running"' && break
sleep 30
done---
5. End-to-End Execution Flow (Summary)
1. Initialize: verify credentials -> GetCallerIdentity -> confirm region
2. Understand the task: is the user querying or making changes?
3. Locate the API: ve --help first -> Python helpers as fallback
4. Query dependent resources: use Describe*/List* to obtain required IDs
5. Read operation -> execute directly and display results
Write operation -> show command -> DryRun (if supported) -> user confirmation -> execute
6. Parse the response and report results to the user---
6. Service-Specific Notes
Consult or update the corresponding notes file when encountering service-specific issues:
- ECS: references/ecs.md
- VPC: references/vpc.md
- CR: references/cr.md
- ALB: references/alb.md
- CLB: references/clb.md
- VKE: references/vke.md
- veFaaS: references/vefaas.md
- RDS: references/rds.md
- Message Queue: references/mq.md
- Storage: references/storage.md
- Observability: references/observability.md
- DNS/Edge: references/dns-edge.md
- IAM: references/iam.md
- KMS: references/kms.md
- Redis: references/redis.md
- NAT Gateway: references/natgateway.md
- EBS: references/ebs.md
- Extension APIs: references/extend-apis.md
ALB Service Notes
Flat Parameter Mode
ALB create APIs use flat CLI parameters, not JSON body mode. Nested arrays use indexed dot notation such as --ZoneMappings.1.ZoneId.
For private ALB smoke tests, keep all EIP/public address fields unset unless public exposure is explicitly under test:
ve alb CreateLoadBalancer \
--RegionId cn-beijing \
--LoadBalancerName cli-skill-test-alb \
--Type private \
--VpcId vpc-xxxx \
--SubnetId subnet-xxxx \
--ZoneMappings.1.ZoneId cn-beijing-b \
--ZoneMappings.1.SubnetId subnet-xxxx \
--LoadBalancerBillingType 1 \
--LoadBalancerEdition Basic \
--Tags.1.Key publish-by \
--Tags.1.Value deploy-skillObserved in cn-beijing: a private Basic ALB created with publish-by=deploy-skill appeared in DescribeLoadBalancers with the tag attached, then DeleteLoadBalancer removed it. ALB creation is billable and depends on real VPC/subnet choices, so keep lifecycle tests short and delete the test load balancer after validation.
CLB Service Notes
Zone and EIP Behavior
DescribeZones returns master/slave zone pairs, not a flat zone list. Do not feed the raw structure into single-zone fields without choosing the intended master/slave relationship.
For private CLB tests, leave EipBillingConfig.* unset. The create API exposes EIP billing fields, so copying a public example can accidentally allocate a public endpoint.
The delete API supports force deletion:
ve clb DeleteLoadBalancer --LoadBalancerId <clb-id> --ForceDelete trueObserved in cn-beijing: CLBs, listeners, server groups, and certificates returned empty lists. No lifecycle test was run because CLB is billable and may allocate dependent resources.
CR Service Notes
Namespace and Repository Calls Need a Registry
ListNamespaces without Registry fails with:
MissingParameter.Registry: The required parameter Registry is missing.Passing a non-existent registry reaches the service and fails with:
NotFound.Registry: The specified resource dummy not found.Use these errors to distinguish JSON/CLI formatting mistakes from a legitimately missing registry.
Write APIs Use JSON Body Mode
CR create APIs use --body JSON mode. Do not create namespaces or repositories until the registry ID/name is known from ListRegistries or CreateRegistry output.
Observed in cn-beijing: ListRegistries returned TotalCount: 0; no CR lifecycle test was run.
Docker Authentication
For push/pull outside the VKE cr-credential-controller passwordless path, authenticate Docker with a short-lived CR authorization token:
token_json=$(ve cr GetAuthorizationToken --Registry "$registry_name")
cr_username=$(printf '%s' "$token_json" | jq -r '.Result.Username // empty')
cr_password=$(printf '%s' "$token_json" | jq -r '.Result.AuthorizationToken')
[ -n "$cr_username" ] || { echo "CR token response missing Result.Username" >&2; exit 1; }
[ -n "$cr_password" ] || { echo "CR token response missing Result.AuthorizationToken" >&2; exit 1; }
printf '%s' "$cr_password" | docker login "$registry_endpoint" \
--username "$cr_username" \
--password-stdinRules:
- Do not print or write
AuthorizationTokento logs. - Re-run
GetAuthorizationTokenwhen Docker push/pull starts failing after a long session; the token is temporary. - If
Usernameis missing from the response, stop and inspect the CR API response; do not invent a fallback username. - For VKE private CR pulls, prefer the
cr-credential-controlleraddon when available; otherwise use an explicit KubernetesimagePullSecret.
DNS and Edge Service Notes
CDN Can Be Disabled at Account Level
ve cdn ListCdnDomains can fail even though the CLI command exists:
OperationDenied.ServiceStopped: 服务处于停用状态,不支持该操作。Treat this as account service state, not as a request-format bug. CDN domain validation requires the service to be enabled first.
DNS, PrivateZone, and WAF Need Real Inputs
ve dns ListZones, ve privatezone ListPrivateZones, and ve waf ListDomain returned successfully in cn-beijing.
Do not use DNS/PrivateZone/WAF creation as a generic smoke test:
- public DNS zones need a real domain ownership/context;
- PrivateZone record tests need an intended VPC binding;
- WAF domain creation needs a real domain/backend/load-balancer context.
EBS Service Notes
Existing Volumes Are Often ECS System Disks
ve storageebs DescribeVolumes returned an existing system volume attached to an ECS instance. Treat every existing volume as a user resource; never detach or delete it during smoke tests.
Explorer Helper Gap
scripts/fetch_swagger.py --service storageebs --list returned HTTP 404 from the Explorer versions endpoint. Use ve storageebs <Action> --help for parameter discovery.
Lifecycle Risk
CreateVolume is billable. If lifecycle testing is approved, use a small postpaid data disk, never attach it to a non-test instance, delete it immediately, and verify the test volume disappears from DescribeVolumes.
ECS Service Notes
Resource Discovery Pitfalls
DescribeInstanceTypes returns only a small default page and does not prove zone inventory. For placement decisions, query the target zone with DescribeAvailableResource; otherwise RunInstances can fail with InvalidInstanceType.NotFound even when the type exists globally.
For instance type inventory, read .Result.AvailableZones[].AvailableResources[] | select(.Type=="InstanceType").SupportedResources[], then keep entries whose Status is Available; do not look for a top-level InstanceTypes list.
veLinux image search should use an exact name prefix. A fuzzy keyword like velinux also matches GPU, Docker, ARM, and other variants. Known useful names:
veLinux 2.0 64veLinux 2.0 ARM 64
Current RunInstances Shape
Current CLI accepts --ZoneId for RunInstances; older examples may show --Placement.ZoneId. Check the installed CLI help before assuming one shape.
RunInstances requires either --Password or --KeyPairName, even when SSH is not opened. For Cloud Assistant-only deployments, generate a one-time strong password and do not log or persist it.
For no-EIP validation, remove all EipAddress.* parameters and use --DryRun true. A successful DryRun exits non-zero and prints DryRunOperation; this is expected and creates no instance.
Inline EIP ChargeType values observed in help/validation are PayByBandwidth, PayByTraffic, and PrePaid.
If creating the VPC/subnet/security group immediately before RunInstances, wait for VPC and security group readiness first. The API can return InvalidVpc.InvalidStatus when a subnet is created right after CreateVpc, and InvalidSecurityGroup.InvalidStatus when an ingress rule is written right after CreateSecurityGroup.
DeleteInstance Status Casing
DescribeInstances can return uppercase statuses such as CREATING and RUNNING. Do not compare only against title-case values.
Calling DeleteInstance while the instance is still CREATING fails with InvalidInstanceStatus. Poll until RUNNING, STOPPED, or another deletable final state before deletion.
Verified no-EIP lifecycle: created a disposable instance, confirmed EipAddress: null, waited for RUNNING, deleted it, and verified a follow-up name query returned TotalCount: 0.
Cloud Assistant Gotchas
After InstallCloudAssistant, the agent can report ReadyReboot; RunCommand may keep timing out until the instance is rebooted. Prefer --InstallRunCommandAgent true during instance creation.
RunCommand requires an explicit --InvocationName. Use a name no longer than 64 characters, containing only Chinese characters, letters, digits, underscores, or hyphens, and do not start it with a digit or hyphen. Keep it short and stable, for example deploy-check; if omitted, some ve CLI versions can derive the invocation name from CommandContent, and base64 or long shell payloads then fail with LimitExceeded.MaximumInvocationName.
RunCommand --Timeout minimum is 60 seconds. Lower values fail with LimitExceeded.MaximumTimeout.
RunCommand --CommandContent must be base64-encoded shell content. Passing plain text such as echo OK can fail with InvalidBase64Content.Malformed.
Treat RunCommand as scheduling only. Poll invocation results and read the result status from .Result.InvocationResults[0].InvocationResultStatus; terminal values include Success, Failed, and Timeout. Pair it with .Result.InvocationResults[0].ExitCode, and decode .Result.InvocationResults[0].Output from base64 when inspecting command output.
command_b64=$(printf '%s' 'systemctl is-active --quiet app && echo OK' | base64 | tr -d '\n')
invocation_id=$(ve ecs RunCommand \
--Type Shell \
--InstanceIds.1 "$instance_id" \
--InvocationName "deploy-check" \
--Timeout 60 \
--CommandContent "$command_b64" \
| jq -r '.Result.InvocationId')
for _ in $(seq 1 30); do
result=$(ve ecs DescribeInvocationResults --InvocationId "$invocation_id" --InstanceId "$instance_id")
result_status=$(printf '%s' "$result" | jq -r '.Result.InvocationResults[0].InvocationResultStatus // empty')
exit_code=$(printf '%s' "$result" | jq -r '.Result.InvocationResults[0].ExitCode // empty')
case "$result_status" in
Success|Failed|Timeout) break ;;
esac
sleep 5
done
echo "$result" \
| jq -r '.Result.InvocationResults[0] | [.InvocationResultStatus, (.ExitCode | tostring)] | @tsv'
if [ "$result_status" != "Success" ] || [ "$exit_code" != "0" ]; then
echo "RunCommand failed or timed out" >&2
exit 1
fi
echo "$result" \
| jq -r '.Result.InvocationResults[0].Output // ""' \
| base64 -dveLinux Docker Deployment
veLinux 2 may report VERSION_CODENAME=lyra; do not use lyra as the Docker official Debian repository codename. For quick deployments, prefer the distribution package:
apt-get update
apt-get install -y docker.io
systemctl enable --now dockerDocker Hub and GHCR can time out from China regions. Prefer Volcengine CR, user-provided registries, or verified mirror pull commands from the mirror's own image detail page. docker.aityp.com can be used as a search/sync candidate for some images, but do not assume docker.aityp.com/<image> is a universal drop-in registry path.
Extended APIs
Use this reference when a Volcengine API is missing from the normal ve command surface but can still be called as a Volcengine OpenAPI extension.
The helper script is:
python3 scripts/call_extend_api.pyIt embeds the extension API registry and the request-signing code needed by these APIs. It does not import external repositories.
Credentials
The script resolves credentials in this order:
1. Environment variables. 2. Volcengine CLI config through the Python SDK CLIConfigCredentialProvider.
If VOLCENGINE_ACCESS_KEY or VOLCENGINE_SECRET_KEY is not detected, the helper prints a notice and then tries to reuse the active ve CLI profile. The SDK provider supports CLI profiles whose mode is ak, ramrolearn, oidc, ecsrole, sso, or console-login.
Environment variables:
export VOLCENGINE_ACCESS_KEY="AK..."
export VOLCENGINE_SECRET_KEY="SK..."
export VOLCENGINE_REGION="cn-beijing"
# Optional:
export VOLCENGINE_SESSION_TOKEN="..."CLI profile fallback:
python3 scripts/call_extend_api.py \
--profile default \
--api QueryMetrics \
--params '{"workspace":"vmp-workspace-id","query":"up"}'Use --config-file /path/to/config.json only when the CLI config is not in the default location. Do not print or echo secrets in the conversation. If both environment credentials and CLI profile resolution fail, ask the user to run ve login, configure a ve profile, or set VOLCENGINE_ACCESS_KEY and VOLCENGINE_SECRET_KEY in their shell.
Discover Supported APIs
List the registry:
python3 scripts/call_extend_api.py --listDescribe one API:
python3 scripts/call_extend_api.py --describe QueryMetricsInclude test-only entries:
python3 scripts/call_extend_api.py --list --include-testCall An API
Basic form:
python3 scripts/call_extend_api.py \
--api QueryMetrics \
--params '{"workspace":"vmp-workspace-id","query":"up"}'Pass params from a file:
python3 scripts/call_extend_api.py \
--api ListPipelineRunStagesInner \
--params @request.jsonAssert the expected method:
python3 scripts/call_extend_api.py \
--api ListAccelerateAreas \
--method GET \
--params '{}'Override endpoint host only when the registry has a product endpoint or the user provides one:
python3 scripts/call_extend_api.py \
--api QueryMetrics \
--host open.volcengineapi.com \
--params '{"workspace":"vmp-workspace-id","query":"up"}'The script resolves service, version, method, default endpoint, scheme, and default content_type from the registry. If --method disagrees with the registry, it fails before making a request. For APIs marked with query parameters in the registry, include those keys in --params; the helper signs them as URL query parameters and sends the remaining keys as the body.
VMP Metric API Notes
The VMP extension APIs mirror Prometheus query APIs but are signed through Volcengine OpenAPI:
QueryMetrics,QueryMetricsRange,GetLabels, andGetSeriesputworkspacein the URL query.GetLabelValuesputs bothworkspaceandlabelin the URL query.- Remaining parameters are sent in an
application/x-www-form-urlencodedbody. - Series selector parameters must use the Prometheus HTTP API name
match[]. Do not passmatchormatchesforGetSeries,GetLabels, orGetLabelValues.
Example:
python3 scripts/call_extend_api.py \
--api GetSeries \
--params '{"workspace":"vmp-workspace-id","match[]":["up{job=\"node\"}"]}'To validate VMP APIs with real data, create a temporary VMP workspace, enable public access, and configure authentication before writing samples:
ve vmp CreateWorkspace --body '{
"Name":"codex-vmp-api-verify",
"InstanceTypeId":"vmp.standard.15d",
"Tags":[{"Key":"publish-by","Value":"deploy-skill"}],
"DeleteProtectionEnabled":false,
"PublicAccessEnabled":true,
"PublicWriteBandwidth":1,
"PublicQueryBandwidth":1
}'Set BasicAuth with Username and a base64-encoded Password. Do not set AuthType to Basic; the service reports InvalidParameter.AuthType. Confirm with GetWorkspaceAuthInfo, which should return AuthType: BasicAuth.
ve vmp UpdateWorkspace --body '{
"Id":"vmp-workspace-id",
"Username":"user-name",
"Password":"base64-encoded-password"
}'Use GetWorkspace to obtain PrometheusWriteEndpoint; append /api/v1/write for Prometheus remote write. A successful remote-write request returns HTTP 204. The path /api/v1/push was not accepted for this workspace during validation.
After writing a sample such as codex_vmp_verify_value{run_id="..."} 42, verify all five extension APIs:
python3 scripts/call_extend_api.py \
--api QueryMetrics \
--params '{"workspace":"vmp-workspace-id","query":"codex_vmp_verify_value{run_id=\"run-id\"}","time":"sample-or-later-unix-second"}'
python3 scripts/call_extend_api.py \
--api QueryMetricsRange \
--params '{"workspace":"vmp-workspace-id","query":"codex_vmp_verify_value{run_id=\"run-id\"}","start":"sample-start","end":"sample-end","step":"10s"}'
python3 scripts/call_extend_api.py \
--api GetLabels \
--params '{"workspace":"vmp-workspace-id","match[]":["codex_vmp_verify_value{run_id=\"run-id\"}"]}'
python3 scripts/call_extend_api.py \
--api GetLabelValues \
--params '{"workspace":"vmp-workspace-id","label":"run_id","match[]":["codex_vmp_verify_value{run_id=\"run-id\"}"]}'
python3 scripts/call_extend_api.py \
--api GetSeries \
--params '{"workspace":"vmp-workspace-id","match[]":["codex_vmp_verify_value{run_id=\"run-id\"}"]}'For instant queries against freshly written remote-write data, query at or after the stored sample timestamp. A query before the sample timestamp can correctly return an empty vector even though range queries already show the sample.
Safety
Apply the same safety rules as normal ve calls:
- Read-only actions such as
Describe*,List*,Get*,Query*,Check*, andSearch*can be run directly when the user asks for them. - Write actions such as
Create*,Run*,Update*,Set*,Start*,Register*, andImport*require confirmation. - Destructive actions such as
Delete*,Stop*, andCancel*require explicit confirmation and an impact summary.
Parameter names and semantics are captured below so this file remains self-contained. Keep params explicit and prefer small read-only calls first.
API Index
Use this index for quick lookup by service or API name.
| Service | APIs |
|---|---|
CDN | DescribeOriginTopStatisticalData |
cp | ListPipelineRunStagesInner |
dcdn | DescribeOriginRealtimeData, DescribeRealtimeData, DescribeTopIPs, DescribeTopReferers, DescribeTopUrls |
domain_openapi | CheckFee, GetAsyncTask, GetDomain, GetTemplate, ListDomains, ListTemplates, RegisterDomain |
flink | CancelGWSApplication, CreateGWSApplicationDraft, DeleteGWSApplication, DeployGWSApplicationDraft, GWSGetEventList, GetGMSProjectDetail, GetGRSAppById, GetGWSApplication, GetGWSApplicationDraft, ListGASLogs, ListGMCSResourcePool, ListGMSProject, ListGWSApplication, ListGWSDirectory, RestartGWSApplication, StartGWSApplication, UpdateGWSApplicationDraft |
ga | DescribeListenerLogs, GetAcceleratorDimension, GetBandwidthPackage, GetBasicEndpointRelatedAccInstanceInfos, GetEndpointRelatedAccInstanceInfos, ListAccelerateAreas, ListBandwidthPackages |
iot | CallService, GetAllLastDevicePropertyValue, GetCustomTopicList, GetDeviceDetail, GetDeviceEventRecordList, GetDeviceList, GetDeviceOverview, GetDeviceServiceCallRecordList, GetDeviceStatus, GetInstanceDetail, GetInstanceEndpoints, GetInstanceList, GetLastDevicePropertyValue, GetProductDetail, GetProductList, GetPropertyValuesByTime, GetThingModel, SetProperty |
live | DescribeLiveBatchStreamSessionData, DescribeLiveBatchStreamTranscodeData |
mcdn | DescribeCdnDomainConfig |
metrics | GetQueryCluster, GetWorkspaceInfo, InfluxQuery, ListPreagg, ListQueryClusters, ListWorkspace, MetricsQuery |
sec_agent | RunAlertFormatter, RunAlertInvestigator, RunDlpScreenshotAnalyzer, RunPcapAnalyzer, RunSensitiveDataDetector, RunThreatIntelProducer, RunWebRiskAssessor |
trademark | GetApplicant, GetRequirement, GetTrademark, ListApplicants, ListBarrierTrademarks, ListRequirements, ListTrademarks, SearchTrademark, SearchTrademarkInfo |
veenedge | GetBandwidthUsage, GetBillingUsageDetail, GetVEENInstanceUsage, GetVEEWInstanceUsage, RebootCloudServer, StartCloudServer, StopCloudServer |
vke | CreateVirtualNode, ListVirtualNodes |
vmp | GetLabelValues, GetLabels, GetSeries, QueryMetrics, QueryMetricsRange |
Supported APIs
| APIName | Service | Version | Method | Purpose |
|---|---|---|---|---|
DescribeOriginTopStatisticalData | CDN | 2021-03-01 | POST | CDN origin-side top statistical data |
ListPipelineRunStagesInner | cp | 2023-05-01 | POST | CodePipeline stage/task list for a pipeline run |
DescribeRealtimeData | dcdn | 2021-04-01 | POST | DCDN realtime edge data |
DescribeOriginRealtimeData | dcdn | 2021-04-01 | POST | DCDN realtime origin data |
DescribeTopIPs | dcdn | 2021-04-01 | POST | DCDN top client IP ranking |
DescribeTopReferers | dcdn | 2021-04-01 | POST | DCDN top referer ranking |
DescribeTopUrls | dcdn | 2021-04-01 | POST | DCDN top URL ranking |
CheckFee, GetDomain, GetAsyncTask, GetTemplate, ListDomains, ListTemplates | domain_openapi | 2022-12-12 | GET | Domain price, domain, task, and template queries |
RegisterDomain | domain_openapi | 2022-12-12 | POST | Register a domain; creates a billable async task |
| Flink GMS/GWS/GAS actions | flink | 2021-06-01, 2022-06-01 | GET/POST | Flink project, resource-pool, draft, application, event, and log operations |
| GA actions | ga | 2022-03-01 | GET/POST | Global Accelerator area, bandwidth package, metric, listener, and endpoint queries |
| IoT actions | iot | 2021-12-14 | POST | IoT instance, product, device, thing-model, service-call, and property operations |
DescribeLiveBatchStreamTranscodeData, DescribeLiveBatchStreamSessionData | live | 2023-01-01 | POST | Live stream transcode/session statistics |
DescribeCdnDomainConfig | mcdn | 2022-03-01 | GET | Multi-cloud CDN domain configuration |
| Metrics actions | metrics | 2024-06-29 | POST | Metrics workspace, query-cluster, pre-aggregation, Influx, and metrics queries |
| Security workflow actions | sec_agent | 2025-01-01 | POST | Security workflow execution and result retrieval |
| Trademark actions | trademark | 2023-06-01 | GET/POST | Trademark applicant, requirement, trademark, barrier, and search queries |
| VEEN actions | veenedge | 2021-04-30 | GET/POST | VEEN instance usage queries and cloud-server start/stop/reboot operations |
CreateVirtualNode, ListVirtualNodes | vke | 2022-05-12 | POST | VKE virtual node create/list |
| VMP metric actions | vmp | 2021-03-03 | POST | Signed Prometheus-compatible query APIs |
Parameter Reference
Unless noted otherwise, pass all fields in --params as one JSON object. The helper resolves Action, Version, Service, Method, endpoint host, and content type from the registry.
CDN
DescribeOriginTopStatisticalData:
- Required:
Domain,StartTime,EndTime,Item,Metric. StartTimeandEndTimeare Unix timestamps in seconds.Itemcurrently supportsurl.Metricsupportsflux,pv,status_2xx,status_3xx,status_4xx,status_5xx.
CodePipeline
ListPipelineRunStagesInner:
- Required:
WorkspaceId,PipelineId,PipelineRunId. - Use it after listing workspaces, pipelines, and pipeline runs with normal
ve cpAPIs.
DCDN
DescribeRealtimeData and DescribeOriginRealtimeData:
- Required:
StartTime,EndTime,Metrics. StartTimeandEndTimeuse"YYYY-MM-DD HH:MM:SS"and the range must be within 24 hours.Metricsis an array. Common values:all,traffic,bandwidth,request,QPS,2xx,3xx,4xx,5xx; edge realtime also supportsRequestHitRateandTrafficHitRate.- Optional filters include
Domains,ProjectName,IspNameEn,RegionNameEn,Protocol,Type,IPVersion.
DescribeTopIPs, DescribeTopReferers, and DescribeTopUrls:
- Required:
StartTime,EndTime,Sort. Sortsupportstraffic,bandwidth,request,QPS.- Optional:
Limit(1-100),ProjectName,Domain,StatusCode.
Domain
CheckFee:
- Required query field:
domain.
GetDomain:
- Optional query fields:
domain,instance_no. Provide at least one useful identifier.
GetAsyncTask:
- Required query field:
task_no.
GetTemplate:
- Required query field:
tag.
ListDomains:
- Optional query fields:
domain,status,verify_status,expired_after,is_auto_renew,domain_name_audit_status,order_by,asc_or_desc,page_number,page_size.
ListTemplates:
- Optional query fields:
registrant_zh,registration_type,tag,status,page_number,page_size.
RegisterDomain:
- Required body fields:
domain,template_tag. - Optional:
period,ns_list,is_auto_renew,package_id. - This is billable and creates an async task. Confirm with the user before calling.
Flink
GET actions:
ListGMSProject: optionalSearchKey,PageSize,PageNum.GetGMSProjectDetail: requiredProjectName.ListGMCSResourcePool:ProjectId, optionalName,NameKey,PageSize,PageNum; use version2022-06-01.GetGRSAppById: requiredAppIdKey.
GWS/GAS POST actions:
ListGWSDirectory: query-style fieldsProjectId,Type(JOBorQUERY).GetGWSApplicationDraft:ProjectId,Id.CreateGWSApplicationDraft:ProjectId,JobName,DirectoryId, optionalJobType,EngineVersion.UpdateGWSApplicationDraft:Id,ProjectId,AccountId,UserId,JobName,JobId,DirectoryId,DirectoryName,SqlText, optionalDynamicOptions,JobType,EngineVersion.DeployGWSApplicationDraft:ProjectId,Id,ResourcePool,Queue, optionalPriority,SchedulePolicy,ScheduleTimeout.DeleteGWSApplication,StartGWSApplication,CancelGWSApplication,RestartGWSApplication:ProjectId,Id; start/restart also supportTypesuch asFROM_NEWorFROM_LATEST.ListGWSApplication: optionalProjectId,JobName,ResourcePool,JobType,State,PageSize,PageNum,SortField,SortOrder.GetGWSApplication: requiredId, optionalAccountId.GWSGetEventList:ProjectId, optionalId,Limit.ListGASLogs:Application,Project,StartTime,EndTime; optionalLevel,Properties.component,Properties.podName,Cursor,PageSize.
Global Accelerator
ListAccelerateAreas:
- No required parameters.
ListBandwidthPackages:
- Required in observed request shape:
BandwidthType,PageNumber,PageSize. - Optional:
AcceleratorId,AccountId,BandwidthPackageId,Domain,Isp,OrderType,State,States,ProjectName,ResourceTagFilter.
GetBandwidthPackage:
- Required:
BandwidthPackageId.
GetAcceleratorDimension:
- Required:
AcceleratorType,TargetName,Filters. Filtersis an array of{ "Name": "...", "Values": ["..."] }.
DescribeListenerLogs:
- Required:
InputIdType,InputId,StartTime,EndTime,Interval. - Optional:
Metrics,RegionType,Region, grouping fields.
GetBasicEndpointRelatedAccInstanceInfos and GetEndpointRelatedAccInstanceInfos:
- Require the endpoint identifier used by the corresponding GA endpoint type. Include pagination fields such as
PageNum/PageSizewhen listing related instances.
IoT
The IoT extension APIs use product-specific identifiers. Common fields are:
- Instance:
InstanceId. - Product:
ProductKeyorProductID. - Device:
DeviceName,DeviceID,DeviceSecret, depending on the API. - Thing model:
ModuleKey,Identifier,PropertyIdentifier,EventIdentifier,ServiceIdentifier. - Pagination and time range:
PageNumber,PageSize,StartTime,EndTime. CallServiceneeds target device identifiers plus service identifier and input params.SetPropertyneeds target device identifiers plus property values.
IoT was not business-tested in this account because the service could not be opened.
Live
DescribeLiveBatchStreamTranscodeData:
- Required:
StartTime,EndTime. - Optional:
DomainList,PageNum,PageSize. - Times are RFC3339 strings, for example
2022-11-10T00:00:00+08:00.
DescribeLiveBatchStreamSessionData:
- Required:
StartTime,EndTime. - Optional:
DomainList,PageNum,PageSize,OnlineUserType, plus stream/session dimensions when available. - Times are RFC3339 strings.
MCDN
DescribeCdnDomainConfig:
- Optional identifiers:
DomainId,DomainName,Vendor,DomainVersion,NormalizeOptions. - Prefer
DomainIdwhen known.
Metrics
ListWorkspace:
- Recommended:
PageNumber,PageSize,ListGlobal. - Optional:
Filters,ProjectName.
GetWorkspaceInfo:
- Required:
WorkspaceId.
ListQueryClusters:
- Recommended:
Pageobject withPageNumberandPageSize. - Optional:
Name,ProjectName.
GetQueryCluster:
- Required:
Id.
ListPreagg:
- Recommended:
PageNumber,PageSize,onlyShowMine. - Optional:
Filters, such as{"WorkspaceName":"..."}.
InfluxQuery and MetricsQuery:
- Require a real workspace/query context and query payload. Include workspace identifier, query expression(s), and time range fields according to the query type.
Security Workflows
RunAlertInvestigator, RunPcapAnalyzer, RunAlertFormatter, RunThreatIntelProducer, RunWebRiskAssessor, RunDlpScreenshotAnalyzer, RunSensitiveDataDetector:
- Require real workflow input such as alert details, PCAP content/reference, URL, screenshot/image data, or text to inspect.
- Do not treat an empty workflow request as validation.
Trademark
GET actions:
GetApplicant: requiredApplicantID.GetTrademark: requiredTrademarkID.GetRequirement: requiredRequirementID.ListApplicants: optionalApplicantType,ApplicantName,Status,Country,PageNumber,PageSize,OrderBy.ListRequirements: optional status/type filters plusPageNumber,PageSize.ListTrademarks: optional trademark/applicant/status filters plusPageNumber,PageSize.ListBarrierTrademarks: optional requirement/trademark identifiers and pagination fields.
POST actions:
SearchTrademarkInfo: requiredClassIDandRegistrationNumber.SearchTrademark: provide at least one useful search condition such asTrademarkName,ApplicantName, orRegistrationNumber; optionalPageNumber,PageSize, class/status filters.
VEEN
StartCloudServer, StopCloudServer, RebootCloudServer:
- Require the target cloud-server identifier. Confirm before start/stop/reboot.
GetVEENInstanceUsage, GetVEEWInstanceUsage, GetBandwidthUsage, GetBillingUsageDetail:
- Usage queries require a billing or resource time range and resource filters. During validation, these GET actions reached
veenedge.volcengineapi.combut returnedMethod Not Allowed.
VKE
ListVirtualNodes:
- Optional pagination and filters such as
PageNumber,PageSize, cluster or virtual-node identifiers.
CreateVirtualNode:
- Required:
Kubeconfig,VirtualNodeConfig. - This creates infrastructure and needs a cleanup plan before execution.
VMP
The VMP action-specific parameter details are in "VMP Metric API Notes" above. The short form:
QueryMetrics: URL query keyworkspace; body fields includequery, optionaltime.QueryMetricsRange: URL query keyworkspace; body fields includequery,start,end,step.GetLabelValues: URL query keysworkspace,label; body supportsmatch[],start,end.GetLabels: URL query keyworkspace; body supportsmatch[],start,end.GetSeries: URL query keyworkspace; body supportsmatch[],start,end.
Notes For Agents
- Prefer normal
vecommands when available. Use this helper only whenvedoes not expose the needed API. - The registry is exact-match by
APIName. If duplicate names appear, pass--service. - Do not invent params. Use the parameter reference in this file; if a required business ID is missing, ask the user for it or locate it with a read-only list/get command.
- Some registry entries have product-specific default hosts such as
cdn.volcengineapi.com,live.volcengineapi.com,iot.cn-shanghai.volcengineapi.com, orveenedge.volcengineapi.com.
Verification Notes
ListPipelineRunStagesInnerwas tested with an existing CP workspace, pipeline, and pipeline run and returned HTTP 200 with stage/task data.ListVirtualNodesreturned HTTP 200 with an empty list.ListPreaggreturned HTTP 200 with an empty response.- VMP metric APIs (
QueryMetrics,QueryMetricsRange,GetLabelValues,GetLabels,GetSeries) were validated against a temporaryvmp.standard.15dworkspace with a real remote-write samplecodex_vmp_verify_value{run_id="run_20260602_1916",source="codex"} 42.QueryMetricsandQueryMetricsRangereturned value42;GetLabelsreturned__name__,run_id, andsource;GetLabelValuesreturned the testrun_id;GetSeriesreturned the complete series. The temporary workspace was deleted afterwards andGetWorkspacereturnedResourceNotExist. - Live batch stream APIs returned permission errors after valid
StartTime/EndTimeparameters were supplied, confirming the earlierInvalidParamwas only parameter-shape related. CreateVirtualNodereached VKE and requiresKubeconfigplusVirtualNodeConfig. It was not executed because it needs an external Kubernetes kubeconfig and the registry does not include a matching virtual-node delete API for cleanup.- Most domain, IoT, trademark, metrics, sec_agent, VEEN, and Flink APIs returned
AccessDeniedfor the tested account. Creating dependent resources cannot bypass IAM denial; use an account with the relevant service permissions for full business-flow testing.
Implementation notes:
| Implementation style | Services observed | Validation note |
|---|---|---|
| Universal client path | cp, vke | Uses the SDK Universal client path with the registry's service, version, method, and content type. |
| Signed Action/Version root path | CDN, dcdn, domain_openapi, ga, live, mcdn, metrics, sec_agent, trademark, veenedge | Uses /?Action=...&Version=... with service-specific signing region, endpoint, scheme, and method from the registry. |
| Signed Prometheus-compatible Action/Version path | vmp | Uses the VMP regional endpoint, URL query keys for workspace/label, and form-encoded body parameters. |
| Flink SDK path | flink | Uses /{Action}/{Version}/{service}/{method}/{content_type} with query parameters for GET actions. |
Current business validation status:
| Service | APIs | Result |
|---|---|---|
vmp | QueryMetrics, QueryMetricsRange, GetLabels, GetLabelValues, GetSeries | Fully validated with a temporary VMP workspace and real remote-write sample value 42. The valid write endpoint was PrometheusWriteEndpoint + /api/v1/write; /api/v1/push was not accepted. Workspace was deleted and deletion was confirmed. |
cp | ListPipelineRunStagesInner | Fully validated with an existing workspace, pipeline, and pipeline run. Response returned real stage/task/step data matching public ve cp ListPipelineRuns context. |
dcdn | DescribeRealtimeData, DescribeOriginRealtimeData, DescribeTopIPs, DescribeTopReferers, DescribeTopUrls | Validated through signed Action/Version calls. StartTime/EndTime must be "YYYY-MM-DD HH:MM:SS" strings. Realtime APIs use Metrics array; top APIs use Sort. Current account has no DCDN domains (ve dcdn ListDomainConfig returned total 0), so statistical responses correctly returned empty result structures. |
domain_openapi | ListDomains, ListTemplates, CheckFee | Validated through signed Action/Version calls. Lists returned empty account resources. CheckFee returned real pricing fields for a sample .com domain, proving business lookup works without creating a domain. Detail/task/register APIs require existing domain/template/task IDs or would register a billable domain. |
trademark | list APIs, SearchTrademark, SearchTrademarkInfo | List APIs returned empty account resources. Search/detail APIs reached business logic: bad/missing params returned field-specific errors; sample public names/registration numbers returned no-match/not-exists business errors. Positive validation needs a known registration number plus class ID accepted by this service. |
metrics | ListWorkspace, ListQueryClusters, ListPreagg | Validated with required fields such as ListGlobal, pagination, and onlyShowMine. Current account has no Metrics workspaces/query clusters/pre-aggregation rules, so responses were HTTP 200 with empty/null business payloads. Get*/query APIs need real workspace or cluster IDs. |
ga | ListAccelerateAreas, ListBandwidthPackages, GetAcceleratorDimension, related endpoint info APIs | ListAccelerateAreas returned real area metadata. Bandwidth/accelerator/dimension/endpoint-related APIs returned empty results, consistent with ve ga ListAccelerators and ve ga ListPublicBandwidthPackages showing zero resources. Resource-detail APIs need GA accelerator, bandwidth package, listener, or endpoint IDs. |
live | DescribeLiveBatchStreamTranscodeData, DescribeLiveBatchStreamSessionData | Validated through signed Action/Version calls and RFC3339 time strings. ListDomainDetail returned no live domains; both private stats APIs returned zero totals and empty stream lists, matching current resource state. |
flink | ListGMSProject, ListGMCSResourcePool | Validated with the Flink SDK path. Both returned valid empty lists. Other GWS/GAS APIs require Flink project/resource pool/application/draft IDs and should not be called without creating those resources. |
vke | ListVirtualNodes, CreateVirtualNode | ListVirtualNodes returned an empty list, consistent with no VKE clusters. CreateVirtualNode reached VKE and returned missing VirtualNodeConfig; real validation requires an existing Kubernetes cluster/kubeconfig and cleanup plan. |
CDN | DescribeOriginTopStatisticalData | Request reached the service, but CDN is stopped/not opened for the account (OperationDenied.ServiceStopped / NotFound.Service). Positive validation requires CDN service activation and a CDN domain with origin traffic. |
mcdn | DescribeCdnDomainConfig | Service is unsubscribed (mcdn.UnsupportedOperation.ServiceUnsubscribed). Positive validation requires MCDN activation and a CDN domain. |
veenedge | GetVEENInstanceUsage, GetVEEWInstanceUsage, GetBandwidthUsage, GetBillingUsageDetail | GET requests against veenedge.volcengineapi.com returned Method Not Allowed. Positive validation requires service-side confirmation of the accepted method. |
sec_agent | Run* workflow APIs | These are workflow/task APIs and need real alert, PCAP, URL, screenshot, or sensitive-text samples to validate meaningful business output. Do not run empty workflow calls as a success criterion. |
iot | all IoT entries | Not tested by user instruction because the service cannot be opened in this account. |
Resource creation gaps for full positive validation:
- Flink: create a temporary Flink project/resource pool/application only after confirming cost/specs, then validate GWS/GAS APIs and delete resources.
- DCDN/CDN/MCDN/Live: require real domains and, for CDN/DCDN stats, actual traffic. Domain ownership, ICP/filing, origin, and certificate setup may be required.
- GA: create accelerator/bandwidth/listener/endpoint resources only after confirming cost and cleanup commands.
- VKE virtual node: requires an existing VKE cluster and kubeconfig; the helper registry does not include a matching virtual-node delete API, so cleanup must be planned through public VKE APIs.
- SecAgent: requires real security workflow input samples and expected-output criteria.
IAM Service Notes
UpdateUser does not support Tags
UpdateUser can only modify basic attributes such as Description and DisplayName. It cannot manage tags.
Correct approach: use TagResources to tag users separately:
ve iam TagResources --ResourceType "User" --ResourceNames.1 "<UserName>" --Tags.1.Key "key" --Tags.1.Value "value"The same applies to roles — use TagResources with ResourceType set to Role.
---
Verified temporary user tag flow
For testing IAM user tagging, create a disposable user and delete it in the same run. CreateUser can set initial tags, and TagResources adds more tags afterward.
user_name="cli-skill-test-user"
ve iam CreateUser \
--UserName "$user_name" \
--Description "cli-skill-test" \
--Tags.1.Key "publish-by" \
--Tags.1.Value "deploy-skill"
ve iam TagResources \
--ResourceType "User" \
--ResourceNames.1 "$user_name" \
--Tags.1.Key "purpose2" \
--Tags.1.Value "cli-skill-test"
ve iam GetUser --UserName "$user_name"
ve iam UntagResources \
--ResourceType "User" \
--ResourceNames.1 "$user_name" \
--TagKeys.1 "purpose2"
ve iam DeleteUser --UserName "$user_name"After deletion, ve iam GetUser --UserName "$user_name" should fail with UserNotExist; use that as the release confirmation for disposable IAM users.
KMS Service Notes
DescribeKeys Requires a Keyring
DescribeKeys requires KeyringName or KeyringID; omitting both returns MissingParameter.
Observed in cn-beijing: DescribeKeyrings returned an existing keyring, DescribeKeys worked when scoped to that keyring, and DescribeSecrets returned TotalCount: 0.
Deletion Is Scheduled
KMS keys and secrets are not immediate-delete resources. The current action list exposes ScheduleKeyDeletion, CancelKeyDeletion, ScheduleSecretDeletion, and CancelSecretDeletion.
Prefer read/help validation unless the test explicitly accepts scheduled deletion and follow-up cleanup tracking. Disposable tests should use a dedicated keyring name prefix.
Message Queue Service Notes
Kafka and RocketMQ Need Pagination
Kafka and RocketMQ DescribeInstances require PageNumber and PageSize; omitting them returns an invalid page parameter error.
ve kafka DescribeInstances --body '{"RegionId":"cn-beijing","PageNumber":1,"PageSize":10}'
ve rocketmq DescribeInstances --body '{"RegionId":"cn-beijing","PageNumber":1,"PageSize":10}'RabbitMQ DescribeInstances worked without the same pagination body in this environment.
Safer Smoke-Test Surface
Kafka/RabbitMQ/RocketMQ allow-list read APIs returned successfully. Allow-list create/delete is a safer lifecycle candidate than full broker instance creation, but still requires explicit approval and final cleanup verification.
Observed in cn-beijing: Kafka, RabbitMQ, RocketMQ, and BMQ instance lists were empty. BMQ reported cn-beijing-b as sold out while other Beijing zones were available.
NAT Gateway Service Notes
Existing NAT Resources Must Be Treated as User-Owned
DescribeNatGateways returned existing account resources in cn-beijing, including EIP-backed SNAT configuration. Do not delete or modify them unless they were created by the current test run.
Observed:
- NAT gateway available zones included
cn-beijing-athroughcn-beijing-d. DescribeSnatEntriesreturned existing SNAT entries.DescribeDnatEntriesreturnedTotalCount: 0.
EIP-Dependent Paths Are Separate
Public NAT SNAT/DNAT flows usually require EIP allocation or association. If EIP allocation is out of scope, validate only read/discovery behavior or use an explicitly approved test-owned EIP.
Observability Service Notes
TLS Is Not Available in This ve Build
The current ve command list does not include tls. Running ve tls --help returns unknown command.
Do not troubleshoot TLS resource operations as CLI parameter mistakes in this environment; there is no matching ve tls command to validate.
CloudMonitor Read Path Works
ve cloudmonitor ListRules returned an empty Data array in cn-beijing. No CloudMonitor rule lifecycle test was run.
RDS Service Notes
Explorer Helper Gap
scripts/fetch_swagger.py --service rdsmysql --list, rdspostgresql, and rdsmssql returned HTTP 404 from the Explorer versions endpoint. Use ve <service> <Action> --help for these service schemas.
Verified CLI service names in the current ve build:
| Engine | CLI service |
|---|---|
| MySQL | rdsmysql (rds_mysql alias also exists) |
| PostgreSQL | rdspostgresql |
| SQL Server | rdsmssql |
Do not use ve rds_postgresql or ve rds_mssql; they return unknown command.
PostgreSQL CLI Pitfalls
CreateDBInstance uses --body JSON. For HA PostgreSQL, include both a Primary and a Secondary item in NodeInfo; a single primary node fails with Secondary Node Number is not equal to 1.
CreateDBAccount accepts AccountPrivileges = "Inherit,Login" for application login accounts. Do not use MySQL/Redis-style ReadWrite for PostgreSQL accounts.
Create the database with Owner set to the app account when possible. If migrations still fail on the default public schema, call ModifySchemaOwner for SchemaName = "public" and the application database/account before running migrations.
For CreateDatabase, omit CharacterSetName unless the accepted enum has been verified for that API call. A real CLI deployment rejected uppercase UTF8; the Terraform provider example uses lowercase utf8.
PostgreSQL instance creation takes several minutes. Even after the instance reports Running, account/database/schema/endpoint operations can briefly fail because the instance is in exclusive status. Retry those follow-up operations with short sleeps instead of recreating the instance.
Lifecycle and Cleanup Risk
RDS instance creation is long-running and billable. Lifecycle tests should create the smallest postpaid instance, wait until the engine-specific status is available, delete it immediately, and verify the engine-specific list/detail API no longer returns it.
Observed in cn-beijing:
- MySQL, PostgreSQL, and SQL Server instance lists returned
0. - MySQL, PostgreSQL, and SQL Server allow-list read APIs returned successfully.
Existing allow lists are account resources and must not be deleted unless created by the current test run.
Redis Service Notes
Swagger/Explorer Gaps
scripts/fetch_swagger.py --service redis --action <Action> can fail with OpenAPI Explorer HTTP 500 for Redis actions including allow-list and instance lifecycle APIs. Fall back to ve redis <Action> --help for the JSON body schema.
Allow-List Cleanup Ordering
Calling DeleteAllowList immediately after DeleteDBInstance can fail with AllowListBindInstanceCannotDelete because the instance has not been fully removed. Wait until DescribeDBInstanceDetail returns not found, then delete the allow list.
DescribeAllowLists requires RegionId and has no useful name filter in CLI help. Passing AllowListName can still return an unfiltered list, so it cannot prove cleanup.
Correct cleanup proof: query the deleted allow-list ID directly and expect AllowListNotExist.
ve redis DescribeAllowListDetail --body '{"AllowListId":"acl-xxx"}'Observed lifecycle note: creating an allow list with AllowListType set to IPv4 returned detail output with AllowListType shown as DualStack; do not use that field alone as an echo check.
Parameter Groups Need Pagination
DescribeParameterGroups requires RegionId, PageNumber, and PageSize. Calling it with only RegionId fails with:
Missing Params: PageNumber,PageSizeObserved in cn-beijing: the paginated call returned system default parameter groups for Redis 4.0, 5.0, 6.0, and 7.0.
Minimal Instance Parameters That Were Easy to Get Wrong
For the disposable Redis instance lifecycle test, the following details mattered:
NoAuthModeusesclose, notdisabled.ConfigureNodesmust include the subnet AZ, for example{"AZ":"cn-beijing-b"}.- Deletion protection must be disabled for disposable tests.
- Poll
DescribeDBInstanceDetailthroughDeletinguntil the detail API returns not found; only then clean dependent allow lists.
Verified lifecycle: created a minimal postpaid Redis 6.0 instance with one 512 MB shard, deleted it, confirmed DescribeDBInstances had no test instance, then deleted the associated allow list and verified AllowListNotExist.
Shell Cleanup Trap
With set -o pipefail, password snippets like tr -dc ... | head -c 14 can exit with code 141 because head closes the pipe. If used inside a cleanup-protected resource test, this can abort after prerequisites are created.
Use a generator that does not rely on an early-closing pipe, or temporarily disable pipefail around password generation. Do not print Redis passwords in logs.
Storage Service Notes
TOS Is Not Available in This ve Build
The current ve command list does not include tos. Running ve tos --help returns unknown command.
Do not troubleshoot TOS bucket operations as CLI parameter mistakes in this environment; there is no matching ve tos command to validate.
Use tosutil for TOS bucket/object operations when it is installed. Verified local tosutil v4.1.4 help exposes:
tosutil mb tos://bucket-name -acl=private -sc=STANDARD
tosutil cp ./dist/app.tar.gz tos://bucket-name/artifacts/app.tar.gz
tosutil presign tos://bucket-name/artifacts/app.tar.gz -vp=15min
tosutil stat tos://bucket-name/artifacts/app.tar.gzIf a deployment path cannot require tosutil, keep it optional and provide SSH/scp or user-provided artifact URL fallback.
File-System Creation Is Billable
EFS, FileNAS, and vePFS read paths worked in cn-beijing; all returned empty filesystem lists.
Creation is billable and may require zone/product sale checks. FileNAS and vePFS zone APIs include sale/status details; inspect those before choosing a zone.
veFaaS Service Notes
Dependent APIs Require FunctionId
ListSandboxes requires FunctionId; a dummy ID reaches the service and returns ResourceNotFound.
CreateTimer, CreateKafkaTrigger, and CreateSandbox all depend on an existing function ID. Validate function creation before testing dependent resources.
Delete order for disposable tests:
DeleteKafkaTrigger / DeleteTimer -> KillSandbox if needed -> DeleteFunctionKeep EnableVpc, TOS mount, NAS mount, and TLS log delivery disabled unless those integrations are specifically under test.
Observed in cn-beijing: ListFunctions returned Total: 0, and availability zones were cn-beijing-a through cn-beijing-d.
VKE Service Notes
Cluster APIs Use JSON Body Mode
CreateCluster and DeleteCluster use --body JSON mode.
ve vke CreateCluster --body '{
"Name": "<cluster-name>",
"ClusterConfig": {"SubnetIds": ["<subnet-id>"]},
"PodsConfig": {
"PodNetworkMode": "Flannel",
"FlannelConfig": {"PodCidrs": ["172.16.0.0/16"]}
},
"ServicesConfig": {"ServiceCidrsv4": ["172.20.0.0/16"]},
"Tags": [{"Key": "publish-by", "Value": "deploy-skill"}]
}'
ve vke DeleteCluster --body '{"Id":"<cluster-id>","Force":true}'Lifecycle Risk
Cluster creation is high-cost and long-running, and can create dependent ECS, network, log, and addon resources. Do not run it as a casual smoke test.
If explicitly approved, record every returned resource ID, use DeleteCluster with an explicit retention/deletion policy, and verify ListClusters no longer returns the test cluster. A newly created cluster can reject deletion while its status is Creating or addon sync is still Progressing; poll ListClusters until the cluster reaches Running/Ok, then delete.
Observed in cn-beijing: clusters, node pools, kubeconfigs, and addons all returned empty lists; supported addon/resource-type discovery worked.
Validation note: a no-node Flannel cluster with Tags: [{"Key":"publish-by","Value":"deploy-skill"}] was created in cn-beijing, the tag appeared in ListClusters, then DeleteCluster removed it after the cluster reached Running/Ok.
Addons That Affect Basic Deployments
ListSupportedAddons returned both core-dns and cr-credential-controller as Unmanaged addons.
core-dnsis the cluster DNS/service-discovery foundation. If workloads cannot resolve Kubernetes service names, checkListAddons/cluster addon state before debugging application DNS.cr-credential-controllersupports passwordless pulls from Volcengine CR. Without it, private CR images may require explicit image pull credentials, and Pods can fail with image pull authentication errors.
VPC Service Notes
Creation Has Short Consistency Windows
CreateVpc can return before the VPC accepts child resources. In a cn-beijing ECS+EIP smoke test, creating a subnet immediately after CreateVpc failed once with:
InvalidVpc.InvalidStatus: The specified VPC is not in the correct status for the request.Poll DescribeVpcs --VpcIds.1 "$vpc_id" until .Result.Vpcs[0].Status == "Available" before creating subnets or security groups. Similarly, CreateSecurityGroup can return before ingress rules are accepted; wait until DescribeSecurityGroups --SecurityGroupIds.1 "$sg_id" returns the group before AuthorizeSecurityGroupIngress, or retry InvalidSecurityGroup.InvalidStatus.
Security Group Name Filter Is Indexed
DescribeSecurityGroups does not accept --SecurityGroupName. Passing it is ignored by the CLI because it is not in help output, so the command returns an unfiltered page and cannot prove cleanup.
Use --SecurityGroupNames.1 instead:
ve vpc DescribeSecurityGroups --SecurityGroupNames.1 "cli-skill-test-sg"After deleting a test security group, confirm cleanup with the same indexed filter and require TotalCount: 0.
Security Group Operations Are Async
CreateSecurityGroup, AuthorizeSecurityGroupIngress, RevokeSecurityGroupIngress, and DeleteSecurityGroup return AsyncTaskId.
AuthorizeSecurityGroupIngress uses flat parameters such as --PortStart, --PortEnd, --Protocol, and --CidrIp; do not use --SourceCidrIp or Permissions.* for this CLI command.
For small test resources the next operation usually succeeds immediately, but cleanup verification should still query after delete. The default egress all rule is created automatically and does not need to be revoked before deleting the test security group.
Verified lifecycle: created a disposable security group, added/revoked one ingress rule, deleted it, and verified the indexed name filter returned no match.
EIP Safety
DescribeEipAddresses returned existing EIPs attached to NAT gateways in cn-beijing. Treat them as user resources; never release or disassociate an EIP unless it was created by the current test run.
#!/usr/bin/env python3
# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
# SPDX-License-Identifier: MIT
"""
Call selected Volcengine extension APIs.
This helper intentionally keeps the Universal client code inline and does not import
from any external repository.
"""
from __future__ import annotations
import argparse
import datetime
import hashlib
import hmac
import json
import os
import sys
from dataclasses import dataclass
from typing import Any
from urllib import error as urllib_error
from urllib import request as urllib_request
from urllib.parse import quote, urlencode
try:
from volcenginesdkcore import UniversalApi, UniversalInfo, ApiClient, Configuration
except ImportError as exc: # pragma: no cover - depends on user environment
print(
"Missing dependency: volcenginesdkcore. Install the Volcengine Python SDK before using this script.",
file=sys.stderr,
)
raise SystemExit(2) from exc
DEFAULT_REGION = "cn-beijing"
DEFAULT_HOST = "open.volcengineapi.com"
@dataclass(frozen=True)
class ResolvedCredentials:
ak: str
sk: str
session_token: str = ""
provider_name: str = ""
class CredentialResolutionError(RuntimeError):
pass
def create_universal_info(service, action, version="2021-09-01", method="POST", content_type="application/json"):
if content_type is None:
content_type = "application/json"
if method == "GET":
content_type = "text/plain"
return UniversalInfo(
method=method,
service=service,
version=version,
action=action,
content_type=content_type,
)
def create_api_client(ak, sk, session_token="", region=DEFAULT_REGION, host=DEFAULT_HOST, scheme="https"):
config = Configuration()
config.ak = ak
config.sk = sk
config.host = host
config.scheme = scheme
config.region = region
if session_token:
config.session_token = session_token
return UniversalApi(ApiClient(config))
API_REGISTRY: list[dict[str, Any]] = [
{
"name": "DescribeOriginTopStatisticalData",
"service": "CDN",
"version": "2021-03-01",
"method": "POST",
"content_type": "application/json",
"host": "cdn.volcengineapi.com",
"scheme": "https",
"call_style": "action_version",
"summary": "CDN origin-side top statistical data.",
},
{
"name": "ListPipelineRunStagesInner",
"service": "cp",
"version": "2023-05-01",
"method": "POST",
"content_type": "application/json",
"call_style": "universal",
"summary": "CodePipeline internal stage list for a pipeline run, used to inspect failed stages/tasks.",
},
{
"name": "DescribeRealtimeData",
"service": "dcdn",
"version": "2021-04-01",
"method": "POST",
"content_type": "application/json",
"host": "open.volcengineapi.com",
"scheme": "http",
"call_style": "action_version",
"summary": "DCDN realtime edge data.",
},
{
"name": "DescribeOriginRealtimeData",
"service": "dcdn",
"version": "2021-04-01",
"method": "POST",
"content_type": "application/json",
"host": "open.volcengineapi.com",
"scheme": "http",
"call_style": "action_version",
"summary": "DCDN realtime origin data.",
},
{
"name": "DescribeTopIPs",
"service": "dcdn",
"version": "2021-04-01",
"method": "POST",
"content_type": "application/json",
"host": "open.volcengineapi.com",
"scheme": "http",
"call_style": "action_version",
"summary": "DCDN top client IP ranking.",
},
{
"name": "DescribeTopReferers",
"service": "dcdn",
"version": "2021-04-01",
"method": "POST",
"content_type": "application/json",
"host": "open.volcengineapi.com",
"scheme": "http",
"call_style": "action_version",
"summary": "DCDN top referer ranking.",
},
{
"name": "DescribeTopUrls",
"service": "dcdn",
"version": "2021-04-01",
"method": "POST",
"content_type": "application/json",
"host": "open.volcengineapi.com",
"scheme": "http",
"call_style": "action_version",
"summary": "DCDN top URL ranking.",
},
{
"name": "DescribeListenerLogs",
"service": "ga",
"version": "2022-03-01",
"method": "POST",
"content_type": "application/json",
"host": "open.volcengineapi.com",
"scheme": "http",
"call_style": "action_version",
"summary": "Global Accelerator listener logs.",
},
{
"name": "GetAcceleratorDimension",
"service": "ga",
"version": "2022-03-01",
"method": "POST",
"content_type": "application/json",
"host": "open.volcengineapi.com",
"scheme": "http",
"call_style": "action_version",
"summary": "Global Accelerator metric dimensions for accelerator resources.",
},
{
"name": "GetBandwidthPackage",
"service": "ga",
"version": "2022-03-01",
"method": "POST",
"content_type": "application/json",
"host": "open.volcengineapi.com",
"scheme": "http",
"call_style": "action_version",
"summary": "Global Accelerator bandwidth package detail.",
},
{
"name": "GetBasicEndpointRelatedAccInstanceInfos",
"service": "ga",
"version": "2022-03-01",
"method": "POST",
"content_type": "application/json",
"host": "open.volcengineapi.com",
"scheme": "http",
"call_style": "action_version",
"summary": "Global Accelerator related basic accelerator instance info for an endpoint.",
},
{
"name": "GetEndpointRelatedAccInstanceInfos",
"service": "ga",
"version": "2022-03-01",
"method": "POST",
"content_type": "application/json",
"host": "open.volcengineapi.com",
"scheme": "http",
"call_style": "action_version",
"summary": "Global Accelerator related accelerator instance info for an endpoint.",
},
{
"name": "ListAccelerateAreas",
"service": "ga",
"version": "2022-03-01",
"method": "GET",
"content_type": "text/plain",
"host": "open.volcengineapi.com",
"scheme": "http",
"call_style": "action_version",
"summary": "List Global Accelerator acceleration areas.",
},
{
"name": "ListBandwidthPackages",
"service": "ga",
"version": "2022-03-01",
"method": "POST",
"content_type": "application/json",
"host": "open.volcengineapi.com",
"scheme": "http",
"call_style": "action_version",
"summary": "List Global Accelerator bandwidth packages.",
},
{
"name": "DescribeLiveBatchStreamTranscodeData",
"service": "live",
"version": "2023-01-01",
"method": "POST",
"content_type": "application/json",
"host": "live.volcengineapi.com",
"scheme": "https",
"call_style": "action_version",
"summary": "Live batch stream transcode data.",
},
{
"name": "DescribeLiveBatchStreamSessionData",
"service": "live",
"version": "2023-01-01",
"method": "POST",
"content_type": "application/json",
"host": "live.volcengineapi.com",
"scheme": "https",
"call_style": "action_version",
"summary": "Live batch stream session data.",
},
{
"name": "DescribeCdnDomainConfig",
"service": "mcdn",
"version": "2022-03-01",
"method": "GET",
"content_type": "text/plain",
"host": "open.volcengineapi.com",
"scheme": "http",
"call_style": "action_version",
"summary": "MCDN CDN domain configuration.",
},
{
"name": "CreateVirtualNode",
"service": "vke",
"version": "2022-05-12",
"method": "POST",
"content_type": "application/json",
"call_style": "universal",
"summary": "Create a VKE virtual node.",
},
{
"name": "ListVirtualNodes",
"service": "vke",
"version": "2022-05-12",
"method": "POST",
"content_type": "application/json",
"call_style": "universal",
"summary": "List VKE virtual nodes.",
},
{
"name": "QueryMetrics",
"service": "vmp",
"version": "2021-03-03",
"method": "POST",
"content_type": "application/x-www-form-urlencoded",
"query_keys": ["workspace"],
"host_template": "vmp.{region}.volcengineapi.com",
"scheme": "https",
"summary": "VMP instant PromQL query for a workspace.",
},
{
"name": "QueryMetricsRange",
"service": "vmp",
"version": "2021-03-03",
"method": "POST",
"content_type": "application/x-www-form-urlencoded",
"query_keys": ["workspace"],
"host_template": "vmp.{region}.volcengineapi.com",
"scheme": "https",
"summary": "VMP range PromQL query for a workspace.",
},
{
"name": "GetLabelValues",
"service": "vmp",
"version": "2021-03-03",
"method": "POST",
"content_type": "application/x-www-form-urlencoded",
"query_keys": ["workspace", "label"],
"host_template": "vmp.{region}.volcengineapi.com",
"scheme": "https",
"summary": "VMP label values query.",
},
{
"name": "GetLabels",
"service": "vmp",
"version": "2021-03-03",
"method": "POST",
"content_type": "application/x-www-form-urlencoded",
"query_keys": ["workspace"],
"host_template": "vmp.{region}.volcengineapi.com",
"scheme": "https",
"summary": "VMP label names query.",
},
{
"name": "GetSeries",
"service": "vmp",
"version": "2021-03-03",
"method": "POST",
"content_type": "application/x-www-form-urlencoded",
"query_keys": ["workspace"],
"host_template": "vmp.{region}.volcengineapi.com",
"scheme": "https",
"summary": "VMP series query.",
},
]
def add_actions(
names: list[str],
*,
service: str,
version: str,
method: str,
summary_prefix: str,
content_type: str = "application/json",
host: str | None = None,
host_template: str | None = None,
scheme: str | None = None,
call_style: str = "action_version",
query_keys: list[str] | None = None,
preserve_query_keys_in_body: list[str] | None = None,
test_only: bool = False,
) -> None:
for name in names:
entry = {
"name": name,
"service": service,
"version": version,
"method": method,
"content_type": content_type,
"call_style": call_style,
"summary": f"{summary_prefix}: {name}.",
}
if host:
entry["host"] = host
if host_template:
entry["host_template"] = host_template
if scheme:
entry["scheme"] = scheme
if query_keys:
entry["query_keys"] = query_keys
if preserve_query_keys_in_body:
entry["preserve_query_keys_in_body"] = preserve_query_keys_in_body
if test_only:
entry["test_only"] = True
API_REGISTRY.append(entry)
add_actions(
[
"RunAlertInvestigator",
"RunPcapAnalyzer",
"RunAlertFormatter",
"RunThreatIntelProducer",
"RunWebRiskAssessor",
"RunDlpScreenshotAnalyzer",
"RunSensitiveDataDetector",
],
service="sec_agent",
version="2025-01-01",
method="POST",
summary_prefix="Security intelligent workflow",
host="open.volcengineapi.com",
scheme="https",
)
add_actions(
["CheckFee", "GetDomain", "GetAsyncTask", "GetTemplate", "ListDomains", "ListTemplates"],
service="domain_openapi",
version="2022-12-12",
method="GET",
summary_prefix="Domain service",
content_type="text/plain",
host="open.volcengineapi.com",
scheme="http",
)
add_actions(
["RegisterDomain"],
service="domain_openapi",
version="2022-12-12",
method="POST",
summary_prefix="Domain service",
host="open.volcengineapi.com",
scheme="http",
)
add_actions(
[
"CallService",
"GetAllLastDevicePropertyValue",
"GetCustomTopicList",
"GetDeviceDetail",
"GetDeviceEventRecordList",
"GetDeviceList",
"GetDeviceOverview",
"GetDeviceStatus",
"GetDeviceServiceCallRecordList",
"GetInstanceDetail",
"GetInstanceEndpoints",
"GetInstanceList",
"GetLastDevicePropertyValue",
"GetProductList",
"GetProductDetail",
"GetPropertyValuesByTime",
"GetThingModel",
"SetProperty",
],
service="iot",
version="2021-12-14",
method="POST",
summary_prefix="IoT device or instance operation",
host="iot.cn-shanghai.volcengineapi.com",
scheme="https",
)
add_actions(
["GetApplicant", "GetTrademark", "ListApplicants", "GetRequirement", "ListRequirements", "ListTrademarks", "ListBarrierTrademarks"],
service="trademark",
version="2023-06-01",
method="GET",
summary_prefix="Trademark query",
content_type="text/plain",
host="open.volcengineapi.com",
scheme="http",
)
add_actions(
["SearchTrademarkInfo", "SearchTrademark"],
service="trademark",
version="2023-06-01",
method="POST",
summary_prefix="Trademark search",
host="open.volcengineapi.com",
scheme="http",
)
add_actions(
["ListWorkspace", "GetWorkspaceInfo", "ListQueryClusters", "GetQueryCluster", "ListPreagg", "InfluxQuery", "MetricsQuery"],
service="metrics",
version="2024-06-29",
method="POST",
summary_prefix="Volcengine Metrics service",
host_template="metrics.{region}.volcengineapi.com",
scheme="https",
)
add_actions(
[
"StartCloudServer",
"StopCloudServer",
"RebootCloudServer",
],
service="veenedge",
version="2021-04-30",
method="POST",
summary_prefix="VEEN edge cloud mutation",
host="veenedge.volcengineapi.com",
)
add_actions(
[
"GetVEENInstanceUsage",
"GetVEEWInstanceUsage",
"GetBandwidthUsage",
"GetBillingUsageDetail",
],
service="veenedge",
version="2021-04-30",
method="GET",
summary_prefix="VEEN edge cloud query",
content_type="text/plain",
host="veenedge.volcengineapi.com",
)
add_actions(
["ListGMSProject", "GetGMSProjectDetail", "GetGRSAppById"],
service="flink",
version="2021-06-01",
method="GET",
summary_prefix="Flink management query",
content_type="text/plain",
host="open.volcengineapi.com",
scheme="https",
call_style="flink_path",
)
add_actions(
["ListGMCSResourcePool"],
service="flink",
version="2022-06-01",
method="GET",
summary_prefix="Flink management query",
content_type="text/plain",
host="open.volcengineapi.com",
scheme="https",
call_style="flink_path",
)
add_actions(
["ListGASLogs", "GetGWSApplication"],
service="flink",
version="2021-06-01",
method="POST",
summary_prefix="Flink GWS/GAS operation",
host="open.volcengineapi.com",
scheme="https",
call_style="flink_path",
)
add_actions(
["ListGWSDirectory"],
service="flink",
version="2021-06-01",
method="POST",
summary_prefix="Flink GWS/GAS operation",
host="open.volcengineapi.com",
scheme="https",
call_style="flink_path",
query_keys=["ProjectId", "Type"],
)
add_actions(
["GetGWSApplicationDraft", "DeleteGWSApplication", "GWSGetEventList", "StartGWSApplication", "CancelGWSApplication", "RestartGWSApplication"],
service="flink",
version="2021-06-01",
method="POST",
summary_prefix="Flink GWS/GAS operation",
host="open.volcengineapi.com",
scheme="https",
call_style="flink_path",
query_keys=["ProjectId"],
)
add_actions(
["CreateGWSApplicationDraft", "UpdateGWSApplicationDraft"],
service="flink",
version="2021-06-01",
method="POST",
summary_prefix="Flink GWS/GAS operation",
host="open.volcengineapi.com",
scheme="https",
call_style="flink_path",
query_keys=["ProjectId"],
preserve_query_keys_in_body=["ProjectId"],
)
add_actions(
["DeployGWSApplicationDraft"],
service="flink",
version="2021-06-01",
method="POST",
summary_prefix="Flink GWS/GAS operation",
host="open.volcengineapi.com",
scheme="https",
call_style="flink_path",
query_keys=["ProjectId", "Id"],
)
add_actions(
["ListGWSApplication"],
service="flink",
version="2021-06-01",
method="POST",
summary_prefix="Flink GWS/GAS operation",
host="open.volcengineapi.com",
scheme="https",
call_style="flink_path",
query_keys=["PageSize", "PageNum", "SortField", "SortOrder"],
)
add_actions(
["GetGMSUserToken"],
service="flink",
version="2021-06-01",
method="GET",
summary_prefix="Flink test-only token query",
content_type="text/plain",
host="open.volcengineapi.com",
scheme="https",
call_style="flink_path",
test_only=True,
)
_REGION_BY_SERVICE = {
"CDN": "cn-north-1",
"dcdn": "cn-north-1",
"ga": "cn-north-1",
"live": "cn-north-1",
"mcdn": "cn-north-1",
"domain_openapi": "cn-north-1",
"trademark": "cn-north-1",
"veenedge": "cn-north-1",
"iot": "cn-shanghai",
}
for _entry in API_REGISTRY:
_region = _REGION_BY_SERVICE.get(_entry["service"])
if _region:
_entry["region"] = _region
def parse_json_value(raw: str | None) -> dict[str, Any]:
if raw is None or raw == "":
return {}
if raw.startswith("@"):
with open(raw[1:], "r", encoding="utf-8") as f:
raw = f.read()
try:
value = json.loads(raw)
except json.JSONDecodeError as exc:
raise SystemExit(f"Invalid JSON for --params: {exc}") from exc
if not isinstance(value, dict):
raise SystemExit("--params must be a JSON object")
return value
def env(name: str, default: str = "") -> str:
return os.getenv(name, default)
def _credential_attr(credentials: Any, attr: str) -> str:
if isinstance(credentials, dict):
value = credentials.get(attr)
else:
value = getattr(credentials, attr, None)
if value is None:
return ""
return str(value).strip()
def _env_value(env_getter, name: str) -> str:
try:
value = env_getter(name, "")
except TypeError:
value = env_getter(name)
if value is None:
return ""
return str(value).strip()
def resolve_volcengine_credentials(
*,
profile: str | None = None,
config_file: str | None = None,
session_token: str | None = None,
env_getter=env,
cli_provider_factory=None,
notify=None,
) -> ResolvedCredentials:
"""Resolve AK/SK from env first, then from the Volcengine CLI credential provider.
CLIConfigCredentialProvider handles the CLI profile modes supported by the
SDK: ak, ramrolearn, oidc, ecsrole, sso, and console-login.
"""
env_ak = _env_value(env_getter, "VOLCENGINE_ACCESS_KEY")
env_sk = _env_value(env_getter, "VOLCENGINE_SECRET_KEY")
if env_ak and env_sk:
return ResolvedCredentials(
ak=env_ak,
sk=env_sk,
session_token=(
str(session_token).strip()
if session_token is not None
else _env_value(env_getter, "VOLCENGINE_SESSION_TOKEN")
),
provider_name="EnvironmentVariableCredentialProvider",
)
if notify:
missing = []
if not env_ak:
missing.append("VOLCENGINE_ACCESS_KEY")
if not env_sk:
missing.append("VOLCENGINE_SECRET_KEY")
notify(
"{} not detected; trying Volcengine CLI credentials from ve login/profile.".format(
" and ".join(missing)
)
)
if cli_provider_factory is None:
try:
from volcenginesdkcore.auth.providers.cli_config_provider import CLIConfigCredentialProvider
except ImportError as exc:
raise CredentialResolutionError(
"VOLCENGINE_ACCESS_KEY/VOLCENGINE_SECRET_KEY are not set, and the installed "
"volcenginesdkcore cannot load CLIConfigCredentialProvider. Install or upgrade "
"the Volcengine Python SDK, run ve login/configure, or set AK/SK environment variables."
) from exc
cli_provider_factory = CLIConfigCredentialProvider
try:
provider = cli_provider_factory(profile_name=profile, config_path=config_file)
credentials = provider.get_credentials()
except Exception as exc:
raise CredentialResolutionError(
"VOLCENGINE_ACCESS_KEY/VOLCENGINE_SECRET_KEY are not set, and Volcengine CLI "
"credential resolution failed. Run ve login, configure a ve profile, or set "
"VOLCENGINE_ACCESS_KEY and VOLCENGINE_SECRET_KEY. Underlying error: {}".format(exc)
) from exc
ak = _credential_attr(credentials, "ak")
sk = _credential_attr(credentials, "sk")
if not ak or not sk:
provider_name = _credential_attr(credentials, "provider_name") or "Volcengine CLI credential provider"
raise CredentialResolutionError(
"{} returned incomplete credentials. Run ve login, configure a ve profile, or set "
"VOLCENGINE_ACCESS_KEY and VOLCENGINE_SECRET_KEY.".format(provider_name)
)
return ResolvedCredentials(
ak=ak,
sk=sk,
session_token=(
str(session_token).strip()
if session_token is not None
else _credential_attr(credentials, "session_token")
),
provider_name=_credential_attr(credentials, "provider_name") or "CLIConfigCredentialProvider",
)
def split_query_body(
params: dict[str, Any],
query_keys: list[str] | None,
preserve_query_keys_in_body: list[str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]]:
if not query_keys:
return {}, params
preserve_query_keys_in_body = preserve_query_keys_in_body or []
query: dict[str, Any] = {}
body: dict[str, Any] = {}
for key, value in params.items():
if key in query_keys:
query[key] = value
if key not in query_keys or key in preserve_query_keys_in_body:
body[key] = value
return query, body
def norm_query(params: dict[str, Any]) -> str:
query = ""
for key in sorted(params.keys()):
value = params[key]
if isinstance(value, list):
for item in value:
query += quote(key, safe="-_.~") + "=" + quote(str(item), safe="-_.~") + "&"
else:
query += quote(key, safe="-_.~") + "=" + quote(str(value), safe="-_.~") + "&"
return query[:-1].replace("+", "%20")
def hmac_sha256(key: bytes, content: str) -> bytes:
return hmac.new(key, content.encode("utf-8"), hashlib.sha256).digest()
def hash_sha256(content: str) -> str:
return hashlib.sha256(content.encode("utf-8")).hexdigest()
def utc_now() -> datetime.datetime:
return datetime.datetime.now(datetime.timezone.utc)
def signed_post_with_query(
*,
ak: str,
sk: str,
session_token: str,
region: str,
host: str,
service: str,
version: str,
action: str,
content_type: str,
query: dict[str, Any],
body: dict[str, Any],
scheme: str,
) -> tuple[Any, int, dict[str, str]]:
if content_type == "application/x-www-form-urlencoded":
body_str = urlencode(body, doseq=True)
else:
body_str = json.dumps(body)
request_query = {"Action": action, "Version": version, **query}
x_date = utc_now().strftime("%Y%m%dT%H%M%SZ")
short_x_date = x_date[:8]
x_content_sha256 = hash_sha256(body_str)
signed_headers = "content-type;host;x-content-sha256;x-date"
canonical_request = "\n".join(
[
"POST",
"/",
norm_query(request_query),
"\n".join(
[
"content-type:" + content_type,
"host:" + host,
"x-content-sha256:" + x_content_sha256,
"x-date:" + x_date,
]
),
"",
signed_headers,
x_content_sha256,
]
)
credential_scope = "/".join([short_x_date, region, service, "request"])
string_to_sign = "\n".join(["HMAC-SHA256", x_date, credential_scope, hash_sha256(canonical_request)])
k_date = hmac_sha256(sk.encode("utf-8"), short_x_date)
k_region = hmac_sha256(k_date, region)
k_service = hmac_sha256(k_region, service)
k_signing = hmac_sha256(k_service, "request")
signature = hmac_sha256(k_signing, string_to_sign).hex()
headers = {
"Host": host,
"X-Content-Sha256": x_content_sha256,
"X-Date": x_date,
"Content-Type": content_type,
"Authorization": (
"HMAC-SHA256 Credential="
+ ak
+ "/"
+ credential_scope
+ ", SignedHeaders="
+ signed_headers
+ ", Signature="
+ signature
),
}
if session_token:
headers["x-security-token"] = session_token
url = f"{scheme}://{host}/?{norm_query(request_query)}"
req = urllib_request.Request(
url=url,
data=body_str.encode("utf-8"),
headers=headers,
method="POST",
)
try:
with urllib_request.urlopen(req, timeout=30) as response:
status_code = response.status
response_headers = dict(response.headers.items())
response_text = response.read().decode("utf-8")
except urllib_error.HTTPError as exc:
status_code = exc.code
response_headers = dict(exc.headers.items())
response_text = exc.read().decode("utf-8")
try:
payload = json.loads(response_text)
except json.JSONDecodeError:
payload = response_text
return payload, status_code, response_headers
def signed_action_version_request(
*,
ak: str,
sk: str,
session_token: str,
region: str,
host: str,
service: str,
version: str,
action: str,
method: str,
content_type: str,
query: dict[str, Any],
body: dict[str, Any],
scheme: str,
) -> tuple[Any, int, dict[str, str]]:
method = method.upper()
if method == "GET":
request_query = {"Action": action, "Version": version, **body, **query}
body_str = ""
else:
request_query = {"Action": action, "Version": version, **query}
if content_type == "application/x-www-form-urlencoded":
body_str = urlencode(body, doseq=True)
else:
body_str = json.dumps(body)
x_date = utc_now().strftime("%Y%m%dT%H%M%SZ")
short_x_date = x_date[:8]
x_content_sha256 = hash_sha256(body_str)
signed_headers = "content-type;host;x-content-sha256;x-date"
canonical_request = "\n".join(
[
method,
"/",
norm_query(request_query),
"\n".join(
[
"content-type:" + content_type,
"host:" + host,
"x-content-sha256:" + x_content_sha256,
"x-date:" + x_date,
]
),
"",
signed_headers,
x_content_sha256,
]
)
credential_scope = "/".join([short_x_date, region, service, "request"])
string_to_sign = "\n".join(["HMAC-SHA256", x_date, credential_scope, hash_sha256(canonical_request)])
k_date = hmac_sha256(sk.encode("utf-8"), short_x_date)
k_region = hmac_sha256(k_date, region)
k_service = hmac_sha256(k_region, service)
k_signing = hmac_sha256(k_service, "request")
signature = hmac_sha256(k_signing, string_to_sign).hex()
headers = {
"Host": host,
"X-Content-Sha256": x_content_sha256,
"X-Date": x_date,
"Content-Type": content_type,
"Authorization": (
"HMAC-SHA256 Credential="
+ ak
+ "/"
+ credential_scope
+ ", SignedHeaders="
+ signed_headers
+ ", Signature="
+ signature
),
}
if session_token:
headers["x-security-token"] = session_token
url = f"{scheme}://{host}/?{norm_query(request_query)}"
data = None if method == "GET" else body_str.encode("utf-8")
req = urllib_request.Request(url=url, data=data, headers=headers, method=method)
try:
with urllib_request.urlopen(req, timeout=30) as response:
status_code = response.status
response_headers = dict(response.headers.items())
response_text = response.read().decode("utf-8")
except urllib_error.HTTPError as exc:
status_code = exc.code
response_headers = dict(exc.headers.items())
response_text = exc.read().decode("utf-8")
try:
payload = json.loads(response_text)
except json.JSONDecodeError:
payload = response_text
return payload, status_code, response_headers
def call_flink_path(
*,
ak: str,
sk: str,
session_token: str,
region: str,
host: str,
service: str,
version: str,
action: str,
method: str,
content_type: str,
params: dict[str, Any],
query_keys: list[str] | None,
preserve_query_keys_in_body: list[str] | None,
scheme: str,
) -> tuple[Any, int, dict[str, str]]:
configuration = Configuration()
configuration.host = host
configuration.scheme = scheme
configuration.ak = ak
configuration.sk = sk
configuration.region = region
if session_token:
configuration.session_token = session_token
client = ApiClient(configuration)
headers = {
"Accept": client.select_header_accept(["application/json"]),
"Content-Type": client.select_header_content_type([content_type]),
}
if method == "GET":
query_params = list(params.items())
body_params = {}
else:
query, body = split_query_body(params, query_keys, preserve_query_keys_in_body)
query_params = list(query.items())
body_params = body
path = f"/{action}/{version}/{service}/{method.lower()}/{content_type.lower().replace('/', '_')}"
response = client.call_api(
path,
method,
{},
query_params,
headers,
body=body_params,
post_params=[],
files={},
response_type=object,
auth_settings=["volcengineSign"],
async_req=False,
_return_http_data_only=False,
_preload_content=True,
_request_timeout=None,
collection_formats={},
)
if isinstance(response, tuple) and len(response) == 3:
return response
return response, 200, {}
def response_from_sdk_api_exception(exc: Exception) -> tuple[Any, int, dict[str, str]] | None:
status = getattr(exc, "status", None)
body = getattr(exc, "body", None)
if status is None or body is None:
return None
if isinstance(body, bytes):
body = body.decode("utf-8", errors="replace")
if isinstance(body, str):
try:
payload = json.loads(body)
except json.JSONDecodeError:
payload = body
else:
payload = body
headers = getattr(exc, "headers", None) or {}
return payload, int(status), dict(headers)
def resolve_host(entry: dict[str, Any], region: str, explicit_host: str | None) -> str:
if explicit_host:
return explicit_host
if entry.get("host_template"):
return entry["host_template"].format(region=region)
if entry.get("host"):
return entry["host"]
return env("VOLCENGINE_ENDPOINT") or DEFAULT_HOST
def registry_by_name(include_test: bool = False) -> dict[str, list[dict[str, Any]]]:
index: dict[str, list[dict[str, Any]]] = {}
for entry in API_REGISTRY:
if entry.get("test_only") and not include_test:
continue
index.setdefault(entry["name"], []).append(entry)
return index
def resolve_api(api_name: str, service: str | None, include_test: bool = False) -> dict[str, Any]:
matches = registry_by_name(include_test).get(api_name, [])
if service:
matches = [entry for entry in matches if entry["service"] == service]
if not matches:
raise SystemExit(f"Unknown APIName: {api_name}. Use --list to inspect supported extension APIs.")
if len(matches) > 1:
services = ", ".join(sorted({entry["service"] for entry in matches}))
raise SystemExit(f"APIName {api_name} is ambiguous across services: {services}. Pass --service.")
return matches[0]
def action_kind(action: str) -> str:
destructive_prefixes = ("Delete", "Terminate", "Release", "Revoke", "Modify", "Stop", "Detach", "Cancel")
write_prefixes = ("Create", "Run", "Allocate", "Attach", "Associate", "Authorize", "Update", "Set", "Start", "Register", "Import")
readonly_prefixes = ("Describe", "List", "Get", "Query", "Check", "Search")
if action.startswith(destructive_prefixes):
return "destructive"
if action.startswith(write_prefixes):
return "write"
if action.startswith(readonly_prefixes):
return "read"
return "unknown"
def print_list(include_test: bool = False) -> None:
entries = [entry for entry in API_REGISTRY if include_test or not entry.get("test_only")]
for entry in sorted(entries, key=lambda e: (e["service"], e["name"])):
print(
f"{entry['name']}\t{entry['service']}\t{entry['version']}\t"
f"{entry['method']}\t{entry.get('summary', '')}"
)
def print_describe(entry: dict[str, Any]) -> None:
print(json.dumps(entry, ensure_ascii=False, indent=2))
def call_api(args: argparse.Namespace) -> int:
entry = resolve_api(args.api_name, args.service, include_test=args.include_test)
expected_method = entry["method"].upper()
if args.method and args.method.upper() != expected_method:
raise SystemExit(f"{entry['name']} uses method {expected_method}, not {args.method.upper()}")
params = parse_json_value(args.params)
query_params, body_params = split_query_body(params, entry.get("query_keys"), entry.get("preserve_query_keys_in_body"))
region = args.region or entry.get("region") or env("VOLCENGINE_REGION") or DEFAULT_REGION
host = resolve_host(entry, region, args.host)
scheme = args.scheme or entry.get("scheme") or "https"
content_type = args.content_type or entry.get("content_type") or "application/json"
try:
credentials = resolve_volcengine_credentials(
profile=args.profile,
config_file=args.config_file,
session_token=args.session_token,
notify=lambda message: print(message, file=sys.stderr),
)
except CredentialResolutionError as exc:
raise SystemExit(str(exc)) from exc
ak = credentials.ak
sk = credentials.sk
session_token = credentials.session_token
info = create_universal_info(
service=entry["service"],
action=entry["name"],
version=entry["version"],
method=expected_method,
content_type=content_type,
)
client = create_api_client(
ak=ak,
sk=sk,
session_token=session_token,
region=region,
host=host,
scheme=scheme,
)
if entry.get("call_style") == "flink_path":
response, status_code, response_headers = call_flink_path(
ak=ak,
sk=sk,
session_token=session_token,
region=region,
host=host,
service=entry["service"],
version=entry["version"],
action=entry["name"],
method=expected_method,
content_type=content_type,
params=params,
query_keys=entry.get("query_keys"),
preserve_query_keys_in_body=entry.get("preserve_query_keys_in_body"),
scheme=scheme,
)
elif query_params and expected_method == "POST":
response, status_code, response_headers = signed_post_with_query(
ak=ak,
sk=sk,
session_token=session_token,
region=region,
host=host,
service=entry["service"],
version=entry["version"],
action=entry["name"],
content_type=content_type,
query=query_params,
body=body_params,
scheme=scheme,
)
elif entry.get("call_style") == "action_version":
response, status_code, response_headers = signed_action_version_request(
ak=ak,
sk=sk,
session_token=session_token,
region=region,
host=host,
service=entry["service"],
version=entry["version"],
action=entry["name"],
method=expected_method,
content_type=content_type,
query=query_params,
body=body_params,
scheme=scheme,
)
else:
try:
response, status_code, response_headers = client.do_call_with_http_info(info=info, body=params)
except Exception as exc:
sdk_error_response = response_from_sdk_api_exception(exc)
if sdk_error_response is None:
raise
response, status_code, response_headers = sdk_error_response
if args.output == "json":
print(json.dumps(response, ensure_ascii=False))
else:
print(f"Status Code: {status_code}")
print(json.dumps(response, ensure_ascii=False, indent=2))
if args.show_headers:
print("Response Headers:")
print(json.dumps(dict(response_headers), ensure_ascii=False, indent=2, default=str))
return 0
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Call Volcengine extension APIs")
parser.add_argument("--api", "--api-name", dest="api_name", help="APIName/Action to call")
parser.add_argument("--params", "--param", default="{}", help="JSON object or @file.json, default {}")
parser.add_argument("--method", choices=["GET", "POST", "get", "post"], help="Optional method assertion")
parser.add_argument("--service", help="ServiceCode disambiguator for duplicate API names")
parser.add_argument("--region", help="Request region, default VOLCENGINE_REGION or cn-beijing")
parser.add_argument("--host", help="Override endpoint host; otherwise uses the registry host or VOLCENGINE_ENDPOINT fallback")
parser.add_argument("--scheme", choices=["https", "http"], help="Override endpoint scheme")
parser.add_argument("--content-type", help="Override content type")
parser.add_argument("--session-token", help="Override VOLCENGINE_SESSION_TOKEN")
parser.add_argument("--profile", help="Volcengine CLI profile name for ve login/config credentials")
parser.add_argument(
"--config-file",
help=(
"Volcengine CLI config file path; defaults to "
"VOLCENGINE_CLI_CONFIG_FILE or ~/.volcengine/config.json"
),
)
parser.add_argument("--output", choices=["pretty", "json"], default="pretty")
parser.add_argument("--show-headers", action="store_true")
parser.add_argument("--include-test", action="store_true", help="Include test-only APIs in --list/--describe/calls")
parser.add_argument("--list", action="store_true", help="List supported extension APIs")
parser.add_argument("--describe", metavar="APIName", help="Print registry metadata for an API")
return parser
def main() -> int:
parser = build_parser()
args = parser.parse_args()
if args.list:
print_list(include_test=args.include_test)
return 0
if args.describe:
print_describe(resolve_api(args.describe, args.service, include_test=args.include_test))
return 0
if not args.api_name:
parser.error("--api is required unless --list or --describe is used")
return call_api(args)
if __name__ == "__main__":
raise SystemExit(main())
#!/usr/bin/env python3
# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
# SPDX-License-Identifier: MIT
"""
Fetch Volcengine API Swagger and convert to Markdown documentation.
Usage: python3 fetch_swagger.py --service ecs --action RunInstances [--version 2020-04-01]
"""
import argparse
import json
import sys
import urllib.request
import urllib.error
BASE_URL = "https://api.volcengine.com/api/common/explorer"
def fetch_json(url, allow_404=False):
try:
with urllib.request.urlopen(url, timeout=10) as resp:
return json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
if allow_404 and e.code == 404:
return None
print(f"Error fetching {url}: {e}", file=sys.stderr)
sys.exit(1)
except urllib.error.URLError as e:
print(f"Error fetching {url}: {e}", file=sys.stderr)
sys.exit(1)
def get_all_versions(service_code):
"""Return all versions for a service, IsDefault=1 first."""
url = f"{BASE_URL}/versions?ServiceCode={service_code}"
data = fetch_json(url)
assert data is not None
versions = data.get("Result", {}).get("Versions", [])
if not versions:
print(f"No version found for service: {service_code}", file=sys.stderr)
sys.exit(1)
versions.sort(key=lambda v: 0 if v.get("IsDefault") == 1 else 1)
return [v["Version"] for v in versions]
def get_default_version(service_code):
return get_all_versions(service_code)[0]
def get_swagger(service_code, version, action_name, all_versions=None):
"""Fetch swagger for an action. If not found in given version, try all other versions."""
def _try(ver):
url = (
f"{BASE_URL}/api-swagger"
f"?ServiceCode={service_code}"
f"&Version={ver}"
f"&APIVersion={ver}"
f"&ActionName={action_name}"
)
data = fetch_json(url, allow_404=True)
if data is None:
return None, None
api = data.get("Result", {}).get("Api")
return api, ver
api, matched_ver = _try(version)
if api:
return api, matched_ver
# Fallback: try remaining versions
if all_versions is None:
all_versions = get_all_versions(service_code)
for ver in all_versions:
if ver == version:
continue
api, matched_ver = _try(ver)
if api:
print(f"Note: action '{action_name}' not found in v{version}, using v{matched_ver}", file=sys.stderr)
return api, matched_ver
print(f"No swagger found for {service_code}.{action_name} in any version: {all_versions}", file=sys.stderr)
sys.exit(1)
def resolve_ref(ref_str, schemas):
"""Resolve a $ref like '#/components/schemas/Foo' to its schema dict."""
if ref_str.startswith("#/components/schemas/"):
name = ref_str[len("#/components/schemas/"):]
return name, schemas.get(name, {})
return ref_str, {}
def get_type_str(schema, schemas):
"""Get a human-readable type string from a schema object."""
if "$ref" in schema:
name, _ = resolve_ref(schema["$ref"], schemas)
return f"object({name})"
t = schema.get("type", "string")
if t == "array":
items = schema.get("items", {})
if "$ref" in items:
name, _ = resolve_ref(items["$ref"], schemas)
return f"array[object({name})]"
return f"array[{items.get('type', 'string')}]"
if t == "integer":
fmt = schema.get("format", "")
return "integer" if not fmt else f"integer({fmt})"
if t == "number":
return "number"
return t
def escape_md(text):
"""Escape pipe characters and newlines for markdown table cells."""
if not text:
return ""
text = str(text)
# Collapse multi-line descriptions to single line, keep it readable
text = text.replace("\n", " ").replace("|", "|")
# Trim markdown tip/note blocks
import re
text = re.sub(r":::.*?:::", "", text, flags=re.DOTALL).strip()
# Limit length
if len(text) > 200:
text = text[:197] + "..."
return text
def format_example(example):
if example is None:
return ""
if isinstance(example, list):
return ", ".join(str(e) for e in example[:2])
return str(example)
def build_params_table(params_list):
"""Build a markdown table from a list of param dicts.
The 'required' value can be:
- True or "required" → 必填 (always required)
- "conditional" → 条件 (required only if the optional parent is set)
- False / "" / None → optional
"""
lines = [
"| 参数名 | 类型 | 必填 | 说明 | 示例值 |",
"|--------|------|:----:|------|--------|",
]
for p in params_list:
req = p.get("required")
if req is True or req == "required":
required = "✓"
elif req == "conditional":
required = "条件"
else:
required = ""
lines.append(
f"| `{p['name']}` | {p['type']} | {required} | {escape_md(p.get('description', ''))} | {escape_md(format_example(p.get('example', '')))} |"
)
return "\n".join(lines)
def parse_get_params(parameters, schemas):
"""Parse GET query parameters, separating flat and nested ($ref) params."""
flat = []
nested = [] # list of (param_name, is_array, schema_name, schema_dict)
for p in parameters:
schema = p.get("schema", {})
name = p.get("name", "")
required = p.get("required", False)
description = p.get("description", "")
example = p.get("example")
if "$ref" in schema:
ref_name, ref_schema = resolve_ref(schema["$ref"], schemas)
nested.append((name, False, ref_name, ref_schema, required, description))
elif schema.get("type") == "array" and "$ref" in schema.get("items", {}):
ref_name, ref_schema = resolve_ref(schema["items"]["$ref"], schemas)
nested.append((name, True, ref_name, ref_schema, required, description))
else:
flat.append({
"name": name,
"type": get_type_str(schema, schemas),
"required": required,
"description": description,
"example": example or schema.get("example"),
})
return flat, nested
def parse_nested_schema(schema_name, schema_dict, parent_prefix, is_array, schemas, visited=None, parent_required=True):
"""Recursively parse a schema into a flat list of param dicts with full paths.
parent_required: True iff every ancestor up to the root is required.
A field is marked:
- "required" if parent_required AND it is required in its own schema
- "conditional" if it is required in its own schema but some ancestor is optional
- "" if it is not required in its own schema
"""
if visited is None:
visited = set()
if schema_name in visited:
return []
visited.add(schema_name)
params = []
properties = schema_dict.get("properties", {})
required_list = schema_dict.get("required", [])
sort_order = schema_dict.get("x-sort-params", [])
# Use sort order if provided, otherwise alphabetical
sorted_keys = sort_order if sort_order else sorted(properties.keys())
# Make sure all keys are included (sort_order might be incomplete)
sorted_keys = sorted_keys + [k for k in properties if k not in sorted_keys]
for prop_name in sorted_keys:
if prop_name not in properties:
continue
prop = properties[prop_name]
own_required = prop_name in required_list
effective_required = parent_required and own_required
if effective_required:
req_status = "required"
elif own_required:
req_status = "conditional"
else:
req_status = ""
full_name = f"{parent_prefix}.N.{prop_name}" if is_array else f"{parent_prefix}.{prop_name}"
if "$ref" in prop:
sub_name, sub_schema = resolve_ref(prop["$ref"], schemas)
params.append({
"name": full_name,
"type": f"object",
"required": req_status,
"description": prop.get("description", ""),
"example": prop.get("example"),
})
params.extend(parse_nested_schema(sub_name, sub_schema, full_name, False, schemas, visited, parent_required=effective_required))
elif prop.get("type") == "array" and "$ref" in prop.get("items", {}):
sub_name, sub_schema = resolve_ref(prop["items"]["$ref"], schemas)
params.append({
"name": full_name,
"type": "array[object]",
"required": req_status,
"description": prop.get("description", ""),
"example": prop.get("example"),
})
params.extend(parse_nested_schema(sub_name, sub_schema, full_name, True, schemas, visited, parent_required=effective_required))
else:
params.append({
"name": full_name,
"type": get_type_str(prop, schemas),
"required": req_status,
"description": prop.get("description", ""),
"example": prop.get("example"),
})
visited.remove(schema_name)
return params
def parse_post_body(request_body, schemas):
"""Parse a POST requestBody schema into flat + nested params."""
content = request_body.get("content", {})
schema = content.get("application/json", {}).get("schema", {})
# Handle top-level $ref
if "$ref" in schema:
_, schema = resolve_ref(schema["$ref"], schemas)
flat = []
nested_sections = []
properties = schema.get("properties", {})
required_list = schema.get("required", [])
sort_order = schema.get("x-sort-params", [])
sorted_keys = sort_order + [k for k in properties if k not in sort_order]
for key in sorted_keys:
if key not in properties:
continue
prop = properties[key]
required = key in required_list
description = prop.get("description", "")
example = prop.get("example")
if "$ref" in prop:
ref_name, ref_schema = resolve_ref(prop["$ref"], schemas)
flat.append({
"name": key,
"type": "object",
"required": required,
"description": description,
"example": example,
})
nested_sections.append((key, False, ref_name, ref_schema, required))
elif prop.get("type") == "array" and "$ref" in prop.get("items", {}):
ref_name, ref_schema = resolve_ref(prop["items"]["$ref"], schemas)
flat.append({
"name": key,
"type": "array[object]",
"required": required,
"description": description,
"example": example,
})
nested_sections.append((key, True, ref_name, ref_schema, required))
else:
flat.append({
"name": key,
"type": get_type_str(prop, schemas),
"required": required,
"description": description,
"example": example or prop.get("example"),
})
return flat, nested_sections
def swagger_to_markdown(service_code, action_name, version, api_swagger):
schemas = api_swagger.get("components", {}).get("schemas", {})
paths = api_swagger.get("paths", {})
# Find the operation
method = "GET"
operation = {}
for _, methods in paths.items():
for m, op in methods.items():
method = m.upper()
operation = op
break
break
# API description (from operation summary or first param description)
summary = operation.get("summary", operation.get("description", ""))
lines = []
lines.append(f"# `ve {service_code} {action_name}`")
if summary:
lines.append(f"\n{summary.strip()}")
lines.append(f"\n**服务**: `{service_code}` | **版本**: `{version}` | **方法**: `{method}`")
lines.append(f"\n**CLI 格式**: `ve {service_code} {action_name} [参数...]`")
lines.append("")
if method == "GET":
parameters = operation.get("parameters", [])
flat_params, nested_list = parse_get_params(parameters, schemas)
lines.append("## 请求参数\n")
if flat_params:
lines.append(build_params_table(flat_params))
else:
lines.append("_无独立参数_")
for (param_name, is_array, schema_name, schema_dict, parent_req, description) in nested_list:
type_label = "array[object]" if is_array else "object"
lines.append(f"\n### 嵌套参数:`{param_name}` ({type_label})\n")
if not parent_req:
lines.append("_父参数可选,下表中『条件』= 仅在父参数被设置时必填_\n")
if description and description != param_name:
lines.append(f"{description}\n")
nested_params = parse_nested_schema(
schema_name, schema_dict, param_name, is_array, schemas, parent_required=bool(parent_req)
)
if nested_params:
lines.append(build_params_table(nested_params))
else: # POST
request_body = operation.get("requestBody", {})
if request_body:
flat_params, nested_sections = parse_post_body(request_body, schemas)
lines.append("## 请求参数 (Request Body JSON)\n")
if flat_params:
lines.append(build_params_table(flat_params))
for (key, is_array, ref_name, ref_schema, parent_req) in nested_sections:
type_label = "array[object]" if is_array else "object"
lines.append(f"\n### 嵌套参数:`{key}` ({type_label})\n")
if not parent_req:
lines.append("_父参数可选,下表中『条件』= 仅在父参数被设置时必填_\n")
nested_params = parse_nested_schema(
ref_name, ref_schema, key, is_array, schemas, parent_required=bool(parent_req)
)
if nested_params:
lines.append(build_params_table(nested_params))
else:
params = operation.get("parameters", [])
flat_params, nested_list = parse_get_params(params, schemas)
lines.append("## 请求参数\n")
if flat_params:
lines.append(build_params_table(flat_params))
for (param_name, is_array, schema_name, schema_dict, parent_req, _desc) in nested_list:
type_label = "array[object]" if is_array else "object"
lines.append(f"\n### 嵌套参数:`{param_name}` ({type_label})\n")
if not parent_req:
lines.append("_父参数可选,下表中『条件』= 仅在父参数被设置时必填_\n")
nested_params = parse_nested_schema(
schema_name, schema_dict, param_name, is_array, schemas, parent_required=bool(parent_req)
)
if nested_params:
lines.append(build_params_table(nested_params))
lines.append("\n---")
lines.append(f"_生成自 Volcengine OpenAPI Explorer: {service_code} {action_name} v{version}_")
return "\n".join(lines)
def list_apis(service_code, version):
"""List all available APIs for a service/version."""
url = (
f"{BASE_URL}/apis"
f"?ServiceCode={service_code}"
f"&Version={version}"
f"&APIVersion={version}"
)
data = fetch_json(url)
assert data is not None
groups = data.get("Result", {}).get("Groups", [])
lines = [f"# {service_code} v{version} API 列表\n"]
for group in groups:
group_name = group.get("Name", "")
apis = group.get("Apis", [])
if apis:
lines.append(f"## {group_name}\n")
for api in apis:
action = api.get("Action", "")
name_cn = api.get("NameCn", "")
desc = api.get("Description", "")
lines.append(f"- **{action}** - {name_cn}: {desc[:80]}")
lines.append("")
return "\n".join(lines)
def main():
parser = argparse.ArgumentParser(
description="Fetch Volcengine API Swagger and output as Markdown"
)
parser.add_argument("--service", "-s", required=True, help="Service code, e.g. ecs")
parser.add_argument("--action", "-a", help="API action name, e.g. RunInstances")
parser.add_argument("--version", "-v", help="API version, e.g. 2020-04-01 (auto-detected if omitted)")
parser.add_argument("--list", "-l", action="store_true", help="List all APIs for the service")
args = parser.parse_args()
service_code = args.service
all_versions = get_all_versions(service_code)
version = args.version or all_versions[0]
if args.list or not args.action:
print(list_apis(service_code, version))
return
api_swagger, matched_version = get_swagger(service_code, version, args.action, all_versions)
md = swagger_to_markdown(service_code, args.action, matched_version, api_swagger)
print(md)
if __name__ == "__main__":
main()