8.3 KiB

Raw Permalink Blame History

Skill Proxy Migration Guide (via Gateway)

This document explains how to migrate a skill script from directly calling a third-party API to using DeerFlow Gateway's transparent proxy, with unified billing orchestration (reserve/finalize).

Applicable scenarios:

Async third-party task skills (image/video/audio generation, etc.)
Existing scripts that directly call providers (for example, RunningHub)

1. Migration Goals

The skill no longer calls third-party domains directly.
Real API keys are replaced by an API key marker in scripts being migrated.
All requests go through /api/proxy/{provider}/....
Gateway handles:

API key injection
Idempotent submit deduplication
Billing reserve/finalize orchestration
Query terminal-state detection and settlement

2. Core Principles

Keep provider names stable (for example, runninghub); do not encode model paths in provider names.
Only submit requests should carry X-Idempotency-Key; query requests should not.
Use X-Thread-Id as a common context header whenever available.
Replace API key usage in scripts with marker API_KEY_MARKER = "__API_KEY_MARKER__" at the actual usage location:

If the script uses header auth, replace with Authorization: Bearer __API_KEY_MARKER__.
If the script uses body auth, replace the body apiKey field value with __API_KEY_MARKER__.
Choose header/body based on the script's real provider contract; do not force a single style.

Use shorthand dot-paths in config extraction fields:

Correct: taskId, status, usage.thirdPartyConsumeMoney
Incorrect: $.taskId, '$'.taskId

3. Skill Script Migration Steps

The examples below assume Python + requests.

Step 1: Add gateway config loaders

Add:

load_skill_env(): loads skill-local .env
get_gateway_config(): reads
- DEER_FLOW_GATEWAY_URL (default http://host.docker.internal:8001)
- RUNNINGHUB_PROXY_PROVIDER (default runninghub)

Step 2: Centralize proxy headers

Implement:

build_proxy_headers(include_idempotency: bool = False)
- always sets Content-Type: application/json
- optionally sets X-Thread-Id
- sets X-Idempotency-Key only when include_idempotency=True
- if this script authenticates via header, set Authorization: Bearer __API_KEY_MARKER__

Step 3: Route submit calls through gateway

Replace:

https://www.runninghub.cn/openapi/v2/<model-path>

With:

{gateway}/api/proxy/{provider}/openapi/v2/<model-path>

And use:

headers=build_proxy_headers(include_idempotency=True)

Step 4: Route query calls through gateway

Replace:

https://www.runninghub.cn/openapi/v2/query

With:

{gateway}/api/proxy/{provider}/openapi/v2/query

And use:

headers=build_proxy_headers()

Step 5: Replace third-party API key values with marker (do not keep real keys in scripts)

Replace in the script being migrated:

Any real Authorization: Bearer <real-key> value -> Authorization: Bearer __API_KEY_MARKER__
Any real body apiKey value -> "apiKey": "__API_KEY_MARKER__"

Keep the original auth location (header vs body) unchanged unless provider API requirements changed.

Reason: gateway injects real credentials by replacing marker values at proxy forwarding time.

Step 6: Keep essential error handling

Recommended checks:

response.raise_for_status()
submit fallback when taskId is missing
query loop timeout/failure handling

Step 7: Clean API key instructions from skill.md

After migrating a skill to gateway proxy, remove any user-facing instructions in skill.md that ask users to configure third-party provider keys (for example, RUNNINGHUB_API_KEY).

What to remove from skill.md:

"Set RUNNINGHUB_API_KEY in .env"
"Create an API key on provider platform"
Any step that tells users to pass Authorization: Bearer ...

What to keep/add in skill.md:

Mention that third-party credentials are handled by gateway config
Keep only skill runtime inputs (prompt, output path, optional style/duration)
Optionally mention gateway-related vars if needed by local debugging:
- DEER_FLOW_GATEWAY_URL
- RUNNINGHUB_PROXY_PROVIDER

Suggested replacement sentence:

"This skill uses DeerFlow Gateway third-party proxy. Provider credentials are configured centrally in gateway and are not required in this skill's local .env."

4. Proxy Config Migration (config.yaml)

Configure submit/query routes under third_party_proxy.providers.<provider>.

Example (RunningHub):

third_party_proxy:
  enabled: true
  providers:
    runninghub:
      base_url: https://www.runninghub.cn
      api_key_env: RUNNINGHUB_API_KEY
      timeout_seconds: 30.0
      frozen_amount: 10.0
      frozen_type: 2
      submit_routes:
        - path_pattern: "/openapi/v2/rhart-image/z-image/turbo-lora"
          task_id_jsonpath: "taskId"
          frozen_amount: 0.03
          frozen_type: 2
        - path_pattern: "/openapi/v2/vidu/text-to-video-q3-turbo"
          task_id_jsonpath: "taskId"
          frozen_amount: 11.2
          frozen_type: 2
      query_routes:
        - path_pattern: "/openapi/v2/query"
          request_task_id_jsonpath: "taskId"
          status_jsonpath: "status"
          success_values: ["SUCCESS"]
          failure_values: ["FAILED", "CANCELLED"]
          usage_jsonpath: "usage.thirdPartyConsumeMoney"

Notes:

Provider-level frozen_amount/frozen_type are defaults.
Submit-route values can override defaults per model endpoint.

5. Reusable Function Template

import os
from pathlib import Path

from dotenv import dotenv_values

API_KEY_MARKER = "__API_KEY_MARKER__"


def load_skill_env() -> dict[str, str]:
    """Load skill-local .env values."""
    env_path = Path(__file__).parent / ".env"
    return {
        key: value
        for key, value in dotenv_values(env_path).items()
        if isinstance(key, str) and isinstance(value, str)
    }


def get_gateway_config() -> tuple[str, str]:
    """Get DeerFlow gateway base URL and proxy provider name."""
    env_vars = load_skill_env()
    gateway_url = os.getenv("DEER_FLOW_GATEWAY_URL") or env_vars.get(
        "DEER_FLOW_GATEWAY_URL",
        "http://host.docker.internal:8001",
    )
    provider = os.getenv("RUNNINGHUB_PROXY_PROVIDER") or env_vars.get(
        "RUNNINGHUB_PROXY_PROVIDER",
        "runninghub",
    )
    return gateway_url.rstrip("/"), provider


def build_proxy_headers(*, include_idempotency: bool = False) -> dict[str, str]:
    headers = {"Content-Type": "application/json"}
    thread_id = os.getenv("THREAD_ID")
    if thread_id:
        headers["X-Thread-Id"] = thread_id
    if include_idempotency:
        from uuid import uuid4
        headers["X-Idempotency-Key"] = str(uuid4())
    return headers


# Optional helper when provider requires header auth in script contract:
def build_proxy_headers_with_auth(*, include_idempotency: bool = False) -> dict[str, str]:
  headers = build_proxy_headers(include_idempotency=include_idempotency)
  headers["Authorization"] = f"Bearer {API_KEY_MARKER}"
  return headers

6. Common Pitfalls

6.1 Response contains taskId but extraction fails

Usually caused by wrong config path syntax:

Wrong: $.taskId or '$'.taskId
Right: taskId

6.2 Why query should not include X-Idempotency-Key

Idempotency keys are for submit deduplication (to avoid duplicate task creation). Query requests are polling and should not generate new idempotency keys.

6.3 Sandbox cannot reach gateway

For Docker-based sandbox execution, use:

DEER_FLOW_GATEWAY_URL=http://host.docker.internal:8001

7. Validation Checklist

No direct third-party domain calls remain in the skill script.
The skill script no longer contains real third-party API key values.
Submit uses proxy URL + include_idempotency=True.
Query uses proxy URL + include_idempotency=False.
Config extraction fields use shorthand dot-paths only.
Submit returns taskId, then query reaches RUNNING/SUCCESS.
Gateway logs show submit/query route hits and finalize flow.
skill.md no longer contains instructions to configure third-party API keys.

8. Reference Implementations

skills/public/image-generation/scripts/generate.py
skills/public/video-generation/scripts/generate.py
backend/app/gateway/routers/third_party.py
backend/app/gateway/third_party_proxy/proxy.py
third_party_proxy section in config.yaml

8.3 KiB Raw Permalink Blame History