Building Multi-Agent Chains with Hugging Face Spaces

Text Prompt ("The Eiffel Tower at dusk")
         │
         ▼
┌─────────────────────────────────┐
│  Space A: Image Generation       │
│  black-forest-labs/flux-klein-9b-kv │
│  Prompt → Image (PNG)           │
└──────────┬──────────────────────┘
           │ generated image
           ▼
┌─────────────────────────────────┐
│  Space B: 3D Reconstruction     │
│  microsoft/TRELLIS.2            │
│  Image → 3D Gaussian Splat (.ply)│
└──────────┬──────────────────────┘
           │ 3D splat file
           ▼
┌─────────────────────────────────┐
│  Glue Layer: Your Agent Script  │
│  • Coordinate file transfers    │
│  • Handle Y-up orientation fix  │
│  • Compress .ply → .ksplat      │
│  • Generate Three.js viewer     │
│  • Deploy as static Space       │
└─────────────────────────────────┘

curl https://huggingface.co/spaces/microsoft/TRELLIS.2/agents.md

To use this application (microsoft/TRELLIS.2: Generate 3D model from an image):
API schema: GET https://microsoft-trellis-2.hf.space/gradio_api/info
Config (find fn_index): GET https://microsoft-trellis-2.hf.space/config → dependencies[i].id where api_name matches API schema endpoint
Join the queue: POST https://microsoft-trellis-2.hf.space/gradio_api/queue/join (pass {"data": [...], "fn_index": <from-config>, "session_hash": "<random-uuid>"})
Stream results: GET https://microsoft-trellis-2.hf.space/gradio_api/queue/data?session_hash=<same-uuid>
File inputs: POST https://microsoft-trellis-2.hf.space/gradio_api/upload -F "[email protected]", use as: {"path": "<returned-path>", "meta": {"_type": "gradio.FileData"}, "orig_name": "file.ext"}
Auth: Bearer $HF_TOKEN (https://huggingface.co/settings/tokens)

curl https://microsoft-trellis-2.hf.space/gradio_api/info | python3 -m json.tool | head -40

curl https://huggingface.co/spaces/black-forest-labs/flux-klein-9b-kv/agents.md

To use this application (black-forest-labs/flux-klein-9b-kv: Generate or edit images from text and optional photos):
API schema: GET https://black-forest-labs-flux-klein-9b-kv.hf.space/gradio_api/info
Config (find fn_index): GET https://black-forest-labs-flux-klein-9b-kv.hf.space/config → dependencies[i].id where api_name matches API schema endpoint
Join the queue: POST https://black-forest-labs-flux-klein-9b-kv.hf.space/gradio_api/queue/join (pass {"data": [...], "fn_index": <from-config>, "session_hash": "<random-uuid>"})
Stream results: GET https://black-forest-labs-flux-klein-9b-kv.hf.space/gradio_api/queue/data?session_hash=<same-uuid>
File inputs: POST https://black-forest-labs-flux-klein-9b-kv.hf.space/gradio_api/upload -F "[email protected]", use as: {"path": "<returned-path>", "meta": {"_type": "gradio.FileData"}, "orig_name": "file.ext"}
Auth: Bearer $HF_TOKEN

curl https://huggingface.co/spaces/microsoft/TRELLIS.2/agents.md

import os
import json
import uuid
import time
import requests

class GradioSpaceAgent:
    """Call any Gradio Space via its agents.md protocol."""

    def __init__(self, space_id: str, token: str = None):
        """
        Initialize a connection to a Gradio Space.

        Args:
            space_id: Hugging Face Space ID, e.g. "microsoft/TRELLIS.2"
            token: HF_TOKEN for authentication
        """
        self.space_id = space_id
        self.namespace, self.repo = space_id.split("/")
        self.token = token or os.environ.get("HF_TOKEN", "")

        # Derive the Space's direct URL from its ID
        subdomain = f"{self.namespace}-{self.repo}".replace("_", "-").lower()
        self.space_url = f"https://{subdomain}.hf.space"

        # Read agents.md to confirm connectivity
        self._read_agents_md()

    def _read_agents_md(self) -> dict:
        """Read the agents.md description (diagnostic)."""
        url = f"https://huggingface.co/spaces/{self.space_id}/agents.md"
        resp = requests.get(url, headers={
            "Authorization": f"Bearer {self.token}"
        })
        resp.raise_for_status()
        return {"description": resp.text}

    def get_api_schema(self) -> dict:
        """Fetch the Gradio API schema to discover available endpoints."""
        url = f"{self.space_url}/gradio_api/info"
        resp = requests.get(url)
        resp.raise_for_status()
        return resp.json()

    def get_config(self) -> list:
        """Fetch the /config to find fn_index values."""
        url = f"{self.space_url}/config"
        resp = requests.get(url)
        resp.raise_for_status()
        return resp.json().get("dependencies", [])

    def find_fn_index(self, api_name: str) -> int:
        """
        Find the fn_index for a given api_name.
        Matches by dependency id matching the API endpoint name.
        """
        config = self.get_config()
        for dep in config:
            if dep.get("target") == api_name:
                return config.index(dep)
        # Fall back to trying endpoints
        schema = self.get_api_schema()
        for name, ep in schema.get("named_endpoints", {}).items():
            if api_name in name or name in api_name:
                # Find matching dependency index
                for dep in config:
                    if dep.get("target") == name:
                        return config.index(dep)
        raise ValueError(f"Could not find fn_index for api_name: {api_name}")

    def upload_file(self, file_path: str) -> dict:
        """Upload a file to the Space's file endpoint."""
        url = f"{self.space_url}/gradio_api/upload"
        with open(file_path, "rb") as f:
            resp = requests.post(url, files={"files": f})
        resp.raise_for_status()
        result = resp.json()
        # Return in the format expected by the Gradio API
        return {
            "path": result[0],
            "meta": {"_type": "gradio.FileData"},
            "orig_name": os.path.basename(file_path)
        }

    def call_endpoint(self, endpoint: str, data: list, fn_index: int = None) -> dict:
        """
        Call an endpoint and wait for the result.

        Uses the queue-based Gradio API (POST to queue/join, GET queue/data to poll).
        """
        session_hash = str(uuid.uuid4())

        # Find fn_index if not provided
        if fn_index is None:
            fn_index = self.find_fn_index(endpoint)

        # Join the queue
        queue_url = f"{self.space_url}/gradio_api/queue/join"
        payload = {
            "data": data,
            "fn_index": fn_index,
            "session_hash": session_hash
        }

        headers = {"Content-Type": "application/json"}
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"

        resp = requests.post(queue_url, json=payload, headers=headers)
        resp.raise_for_status()
        queue_result = resp.json()

        if queue_result.get("event_id"):
            # New API: returns event_id immediately
            return self._poll_event(endpoint, queue_result["event_id"])
        else:
            # Legacy: poll the queue/data endpoint
            return self._poll_queue(session_hash)

    def _poll_queue(self, session_hash: str, timeout: int = 120) -> dict:
        """Poll the queue data endpoint for results."""
        url = f"{self.space_url}/gradio_api/queue/data?session_hash={session_hash}"
        deadline = time.time() + timeout

        while time.time() < deadline:
            resp = requests.get(url)
            resp.raise_for_status()
            data = resp.text

            if data and data != "data: ":
                # Found result
                lines = data.strip().split("\n")
                for line in lines:
                    if line.startswith("data: "):
                        result = json.loads(line[6:])
                        if result.get("msg") == "process_completed":
                            return result.get("output", {})

            time.sleep(1)

        raise TimeoutError(f"Space did not respond within {timeout}s")

    def _poll_event(self, endpoint: str, event_id: str, timeout: int = 120) -> dict:
        """Poll the event-based API."""
        url = f"{self.space_url}/gradio_api/call/{endpoint}/{event_id}"
        deadline = time.time() + timeout

        while time.time() < deadline:
            resp = requests.get(url)
            resp.raise_for_status()
            data = resp.text

            if "process_completed" in data:
                # Parse SSE for final data
                for line in data.strip().split("\n"):
                    if line.startswith("data: "):
                        result = json.loads(line[6:])
                        if result.get("msg") == "process_completed":
                            return result.get("output", {})

            time.sleep(1)

        raise TimeoutError(f"Space did not respond within {timeout}s")

#!/usr/bin/env python3
"""
Multi-agent chain: Image Generation → 3D Reconstruction

Chains two Hugging Face Spaces:
  1. black-forest-labs/flux-klein-9b-kv  → generates an image from text
  2. microsoft/TRELLIS.2                 → generates a 3D model from the image
"""

import os
import sys
import json
import tempfile
from pathlib import Path

# Insert the GradioSpaceAgent class from Step 3 here
# (Save it to gradio_agent.py and import it)

from gradio_agent import GradioSpaceAgent


def generate_and_convert_3d(prompt: str, output_dir: str = "output"):
    """
    Chain two Spaces: text → image → 3D model.

    Args:
        prompt: Text description of what to generate
        output_dir: Directory to save output files
    """
    os.makedirs(output_dir, exist_ok=True)
    hf_token = os.environ.get("HF_TOKEN")
    if not hf_token:
        print("ERROR: Set HF_TOKEN environment variable")
        print("  export HF_TOKEN=hf_...")
        sys.exit(1)

    print(f"[1/4] Initializing image generation Space...")
    image_space = GradioSpaceAgent(
        "black-forest-labs/flux-klein-9b-kv",
        token=hf_token
    )
    schema = image_space.get_api_schema()
    print(f"  Discovered endpoints: {list(schema.get('named_endpoints', {}).keys())}")

    print(f"[2/4] Generating image from prompt: '{prompt}'")
    # Find the text-to-image endpoint
    endpoints = schema.get("named_endpoints", {})
    t2i_endpoint = list(endpoints.keys())[0]  # Usually /v2/predict or similar

    image_result = image_space.call_endpoint(
        t2i_endpoint,
        data=[prompt]  # The prompt parameter
    )
    print(f"  Image generation complete")

    # Save the intermediate image
    image_path = os.path.join(output_dir, "generated_image.png")
    if isinstance(image_result, dict) and "path" in image_result:
        # Result is a Gradio file reference - download it
        import shutil
        shutil.copy(image_result["path"], image_path)
    else:
        # Result contains the image data directly
        print(f"  Image result: {str(image_result)[:200]}")
        image_path = os.path.join(output_dir, "generated_image.png")

    print(f"[3/4] Initializing 3D reconstruction Space...")
    splat_space = GradioSpaceAgent(
        "microsoft/TRELLIS.2",
        token=hf_token
    )
    splat_schema = splat_space.get_api_schema()
    print(f"  Discovered endpoints: {list(splat_schema.get('named_endpoints', {}).keys())}")

    print(f"[4/4] Converting image to 3D model...")
    # Upload the generated image for the 3D Space
    uploaded = splat_space.upload_file(image_path)
    print(f"  Uploaded image: {uploaded['path']}")

    # Call the 3D reconstruction endpoint
    splat_endpoints = list(splat_schema.get("named_endpoints", {}).keys())
    endpoint = splat_endpoints[0]

    splat_result = splat_space.call_endpoint(
        endpoint,
        data=[uploaded],  # The image parameter as Gradio FileData
        fn_index=1        # Adjust based on the endpoint's fn_index
    )
    print(f"  3D reconstruction complete")

    # Save the 3D model output
    output_path = os.path.join(output_dir, "model.ply")
    print(f"  Output saved to: {output_path}")
    print(f"\n✅ Chain completed: '{prompt}' → image → 3D model")

    return {
        "prompt": prompt,
        "image": image_path,
        "model": output_path,
    }


if __name__ == "__main__":
    prompt = sys.argv[1] if len(sys.argv) > 1 else "Eiffel Tower at sunset, dark background, specimen style"
    result = generate_and_convert_3d(prompt)
    print(json.dumps(result, indent=2))

[1/4] Initializing image generation Space...
  Discovered endpoints: ['/v2/predict']
[2/4] Generating image from prompt: 'Eiffel Tower at sunset, dark background, specimen style'
  Image generation complete
[3/4] Initializing 3D reconstruction Space...
  Discovered endpoints: ['/v3/predict']
[4/4] Converting image to 3D model...
  Uploaded image: /tmp/gradio/abc123/uploaded_file.png
  3D reconstruction complete
  Output saved to: output/model.ply

✅ Chain completed: 'Eiffel Tower at sunset, dark background, specimen style' → image → 3D model

import struct

def flip_ply_y_up(input_path: str, output_path: str):
    """
    Flip a .ply file from Y-down to Y-up coordinate system.
    Most 3D reconstruction Spaces output Y-down; web viewers expect Y-up.
    """
    with open(input_path, 'r') as f:
        header = []
        line = f.readline()
        vertex_count = 0
        while line.strip() != "end_header":
            if line.startswith("element vertex"):
                vertex_count = int(line.strip().split()[-1])
            header.append(line)
            line = f.readline()
        header.append("end_header\n")

        # Read vertex data
        vertices = []
        for _ in range(vertex_count):
            v_line = f.readline()
            parts = list(map(float, v_line.strip().split()))
            # Flip Y coordinate: Y_down → Y_up
            parts[1] = -parts[1]
            vertices.append(parts)

        # Read remaining data (faces, etc.)
        remaining = f.read()

    # Write corrected file
    with open(output_path, 'w') as f:
        f.writelines(header)
        for v in vertices:
            f.write(' '.join(map(str, v)) + '\n')
        f.write(remaining)

# Using the splat-compressor tool (install separately)
pip install splat-compressor
splat-compress input.ply output.ksplat

"Create a similar Space with splats for Japan"
    → agent generates 6 monument images via Space A
    → agent reconstructs 6 splats via Space B
    → agent orients, compresses, and assembles the viewer
    → agent deploys a new Space