Python

O SDK oficial de Python do Studio permite executar workflows de forma programática a partir das suas aplicações Python.

O SDK de Python funciona com Python 3.8+ e oferece suporte a execução assíncrona, controle automático de limite de requisições com backoff exponencial e acompanhamento de uso.

Instalação

Instale o SDK com o pip:

pip install studio-sdk

Início rápido

Um exemplo simples para você começar:

from studio import StudioClient

# Initialize the client
client = StudioClient(
    api_key="your-api-key-here",
    base_url="https://agent-studio.seeyu.ai"  # hosted Studio; the default (https://seeyu.ai) is not the API host
)

# Execute a workflow
try:
    result = client.execute_workflow("workflow-id")
    print("Workflow executed successfully:", result)
except Exception as error:
    print("Workflow execution failed:", error)

Referência da API

StudioClient

Construtor

StudioClient(api_key: str, base_url: str = "https://seeyu.ai")

Parâmetros:

api_key (str): sua chave de API do Studio
base_url (str, opcional): URL base da API do Studio. No Studio hospedado, use https://agent-studio.seeyu.ai

Métodos

execute_workflow()

Executa um workflow com dados de entrada opcionais.

result = client.execute_workflow(
    "workflow-id",
    input={"message": "Hello, world!"},
    timeout=30.0  # 30 seconds
)

Parâmetros:

workflow_id (str): o ID do workflow a executar
input (dict, opcional): dados de entrada a passar para o workflow
timeout (float, opcional): tempo limite em segundos (padrão: 30.0)
stream (bool, opcional): habilita respostas em streaming (padrão: False)
selected_outputs (list[str], opcional): saídas de block a transmitir, no formato blockName.attribute (por exemplo, ["agent1.content"])
async_execution (bool, opcional): executa de forma assíncrona (padrão: False)
execution_timeout_seconds (int, opcional): teto opcional, no servidor, para a execução assíncrona, de 1 a 604800 segundos. Requer async_execution=True e não pode estender a política da conta.

Retorna: WorkflowExecutionResult | AsyncExecutionResult

Com async_execution=True, retorna imediatamente com um run_id e uma status_url para consulta. Caso contrário, aguarda a conclusão.

get_workflow_status()

Obtém o status de um workflow (status de deploy etc.).

status = client.get_workflow_status("workflow-id")
print("Is deployed:", status.is_deployed)

Parâmetros:

workflow_id (str): o ID do workflow

Retorna: WorkflowStatus

validate_workflow()

Valida se um workflow está pronto para ser executado.

is_ready = client.validate_workflow("workflow-id")
if is_ready:
    # Workflow is deployed and ready
    pass

Parâmetros:

workflow_id (str): o ID do workflow

Retorna: bool

get_workflow_run()

Obtém o status e, opcionalmente, as saídas de uma execução de workflow.

status = client.get_workflow_run("workflow-id", "run-id", include_output=True)
print("Status:", status["status"])  # 'queued', 'running', 'completed', 'failed'
if status["status"] == "completed":
    print("Output:", status["output"])

Parâmetros:

workflow_id (str): o ID do workflow
run_id (str): o ID da execução retornado pela execução assíncrona
include_output (bool, opcional): inclui a saída final das execuções concluídas
selected_outputs (list[str], opcional): seletores de saída de block a incluir

Retorna: Dict[str, Any]

Campos da resposta:

runId (str): o ID da execução
workflowId (str): o ID do workflow
status (str): um de 'queued', 'pending', 'running', 'paused', 'completed', 'failed', 'cancelled'
startedAt / endedAt (str): timestamps da execução
durationMs (int, opcional): duração em milissegundos
output (any, opcional): a saída do workflow, quando solicitada para uma execução concluída
blockOutputs (dict, opcional): as saídas de block solicitadas
error (dict, opcional): detalhes estruturados da falha, com code, message e, opcionalmente, details

get_job_status()

Obtém o status de um job criado pelo endpoint legado de execução assíncrona. Integrações novas devem usar get_workflow_run() com o ID da execução.

status = client.get_job_status("legacy-job-id")

execute_with_retry()

Executa um workflow com retentativa automática em erros de limite de requisições, usando backoff exponencial.

result = client.execute_with_retry(
    "workflow-id",
    input={"message": "Hello"},
    timeout=30.0,
    max_retries=3,           # Maximum number of retries
    initial_delay=1.0,       # Initial delay in seconds
    max_delay=30.0,          # Maximum delay in seconds
    backoff_multiplier=2.0   # Exponential backoff multiplier
)

Parâmetros:

workflow_id (str): o ID do workflow a executar
input (dict, opcional): dados de entrada a passar para o workflow
timeout (float, opcional): tempo limite em segundos
stream (bool, opcional): habilita respostas em streaming
selected_outputs (list, opcional): saídas de block a transmitir
async_execution (bool, opcional): executa de forma assíncrona
max_retries (int, opcional): número máximo de retentativas (padrão: 3)
initial_delay (float, opcional): espera inicial em segundos (padrão: 1.0)
max_delay (float, opcional): espera máxima em segundos (padrão: 30.0)
backoff_multiplier (float, opcional): multiplicador do backoff (padrão: 2.0)

Retorna: WorkflowExecutionResult | AsyncExecutionResult

A lógica de retentativa usa backoff exponencial (1s → 2s → 4s → 8s...) com ±25% de jitter para evitar o efeito manada. Se a API enviar o header retry-after, esse valor é usado no lugar do cálculo.

get_usage_limits()

Obtém os limites de uso e as informações de cota atuais da sua conta.

limits = client.get_usage_limits()
print("Sync requests remaining:", limits.rate_limit["sync"]["remaining"])
print("Async requests remaining:", limits.rate_limit["async"]["remaining"])
print("Current period cost:", limits.usage["currentPeriodCost"])
print("Plan:", limits.usage["plan"])

Retorna: UsageLimits

Estrutura da resposta:

{
    "success": bool,
    "rateLimit": {
        "sync": {
            "isLimited": bool,
            "limit": int,
            "remaining": int,
            "resetAt": str
        },
        "async": {
            "isLimited": bool,
            "limit": int,
            "remaining": int,
            "resetAt": str
        },
        "authType": str  # 'api' or 'manual'
    },
    "usage": {
        "currentPeriodCost": float,
        "limit": float,
        "plan": str  # e.g., 'free', 'pro'
    }
}

set_api_key()

Atualiza a chave de API.

client.set_api_key("new-api-key")

set_base_url()

Atualiza a URL base.

client.set_base_url("https://my-custom-domain.com")

close()

Fecha a sessão HTTP subjacente.

client.close()

Classes de dados

WorkflowExecutionResult

@dataclass
class WorkflowExecutionResult:
    success: bool
    output: Optional[Any] = None
    error: Optional[str] = None
    logs: Optional[List[Any]] = None
    metadata: Optional[Dict[str, Any]] = None
    trace_spans: Optional[List[Any]] = None
    total_duration: Optional[float] = None
    status: Optional[str] = None

success é True apenas para os status completed e paused. O campo status traz o status terminal do servidor tal como o servidor o informou, então uma execução cancelada (success=False, error=None) é distinguível de uma que falhou.

AsyncExecutionResult

@dataclass
class AsyncExecutionResult:
    success: bool
    run_id: str
    status_url: str
    message: str = ""
    async_execution: bool = True

WorkflowStatus

@dataclass
class WorkflowStatus:
    is_deployed: bool
    deployed_at: Optional[str] = None
    needs_redeployment: bool = False

RateLimitInfo

@dataclass
class RateLimitInfo:
    limit: int
    remaining: int
    reset: int
    retry_after: Optional[int] = None

UsageLimits

@dataclass
class UsageLimits:
    success: bool
    rate_limit: Dict[str, Any]
    usage: Dict[str, Any]

StudioError

class StudioError(Exception):
    def __init__(self, message: str, code: Optional[str] = None, status: Optional[int] = None):
        super().__init__(message)
        self.code = code
        self.status = status

Códigos de erro comuns:

UNAUTHORIZED: chave de API inválida
TIMEOUT: a requisição excedeu o tempo limite
RATE_LIMIT_EXCEEDED: limite de requisições excedido
USAGE_LIMIT_EXCEEDED: limite de uso excedido
EXECUTION_ERROR: a execução do workflow falhou

Exemplos

Execução básica de um workflow

Configure o StudioClient com sua chave de API.

Verifique se o workflow tem deploy e está pronto para ser executado.

Execute o workflow com os seus dados de entrada.

Processe o resultado da execução e trate eventuais erros.

import os
from studio import StudioClient

client = StudioClient(
    api_key=os.getenv("STUDIO_API_KEY"),
    base_url="https://agent-studio.seeyu.ai"
)

def run_workflow():
    try:
        # Check if workflow is ready
        is_ready = client.validate_workflow("my-workflow-id")
        if not is_ready:
            raise Exception("Workflow is not deployed or ready")

        # Execute the workflow
        result = client.execute_workflow(
            "my-workflow-id",
            input={
                "message": "Process this data",
                "user_id": "12345"
            }
        )

        if result.success:
            print("Output:", result.output)
            print("Duration:", result.metadata.get("duration") if result.metadata else None)
        else:
            print("Workflow failed:", result.error)

    except Exception as error:
        print("Error:", error)

run_workflow()

Tratamento de erros

Trate os diferentes tipos de erro que podem acontecer durante a execução de um workflow:

from studio import StudioClient, StudioError
import os

client = StudioClient(
    api_key=os.getenv("STUDIO_API_KEY"),
    base_url="https://agent-studio.seeyu.ai"
)

def execute_with_error_handling():
    try:
        result = client.execute_workflow("workflow-id")
        return result
    except StudioError as error:
        if error.code == "UNAUTHORIZED":
            print("Invalid API key")
        elif error.code == "TIMEOUT":
            print("Workflow execution timed out")
        elif error.code == "USAGE_LIMIT_EXCEEDED":
            print("Usage limit exceeded")
        elif error.code == "INVALID_JSON":
            print("Invalid JSON in request body")
        else:
            print(f"Workflow error: {error}")
        raise
    except Exception as error:
        print(f"Unexpected error: {error}")
        raise

Uso como context manager

Use o cliente como context manager para liberar os recursos automaticamente:

from studio import StudioClient
import os

# Using context manager to automatically close the session
with StudioClient(
    api_key=os.getenv("STUDIO_API_KEY"),
    base_url="https://agent-studio.seeyu.ai"
) as client:
    result = client.execute_workflow("workflow-id")
    print("Result:", result)
# Session is automatically closed here

Execução de workflows em lote

Execute vários workflows de forma eficiente:

from studio import StudioClient
import os

client = StudioClient(
    api_key=os.getenv("STUDIO_API_KEY"),
    base_url="https://agent-studio.seeyu.ai"
)

def execute_workflows_batch(workflow_data_pairs):
    """Execute multiple workflows with different input data."""
    results = []

    for workflow_id, input_data in workflow_data_pairs:
        try:
            # Validate workflow before execution
            if not client.validate_workflow(workflow_id):
                print(f"Skipping {workflow_id}: not deployed")
                continue

            result = client.execute_workflow(workflow_id, input_data)
            results.append({
                "workflow_id": workflow_id,
                "success": result.success,
                "output": result.output,
                "error": result.error
            })

        except Exception as error:
            results.append({
                "workflow_id": workflow_id,
                "success": False,
                "error": str(error)
            })

    return results

# Example usage
workflows = [
    ("workflow-1", {"type": "analysis", "data": "sample1"}),
    ("workflow-2", {"type": "processing", "data": "sample2"}),
]

results = execute_workflows_batch(workflows)
for result in results:
    print(f"Workflow {result['workflow_id']}: {'Success' if result['success'] else 'Failed'}")

Execução assíncrona de workflows

Execute workflows de forma assíncrona para tarefas longas:

import os
import time
from studio import StudioClient

client = StudioClient(
    api_key=os.getenv("STUDIO_API_KEY"),
    base_url="https://agent-studio.seeyu.ai"
)

def execute_async():
    try:
        # Start async execution
        result = client.execute_workflow(
            "workflow-id",
            input={"data": "large dataset"},
            async_execution=True  # Execute asynchronously
        )

        # Check if result is an async execution
        if hasattr(result, 'async_execution') and result.async_execution:
            print(f"Run ID: {result.run_id}")
            print(f"Status endpoint: {result.status_url}")

            # Poll for completion
            status = client.get_workflow_run(
                "workflow-id", result.run_id, include_output=True
            )

            while status["status"] in ["queued", "pending", "running"]:
                print(f"Current status: {status['status']}")
                time.sleep(2)  # Wait 2 seconds
                status = client.get_workflow_run(
                    "workflow-id", result.run_id, include_output=True
                )

            if status["status"] == "completed":
                print("Workflow completed!")
                print(f"Output: {status['output']}")
                print(f"Duration: {status['durationMs']}")
            else:
                print(f"Workflow failed: {status['error']}")

    except Exception as error:
        print(f"Error: {error}")

execute_async()

Limite de requisições e retentativas

Lide com os limites de requisições automaticamente, usando backoff exponencial:

import os
from studio import StudioClient, StudioError

client = StudioClient(
    api_key=os.getenv("STUDIO_API_KEY"),
    base_url="https://agent-studio.seeyu.ai"
)

def execute_with_retry_handling():
    try:
        # Automatically retries on rate limit
        result = client.execute_with_retry(
            "workflow-id",
            input={"message": "Process this"},
            max_retries=5,
            initial_delay=1.0,
            max_delay=60.0,
            backoff_multiplier=2.0
        )

        print(f"Success: {result}")
    except StudioError as error:
        if error.code == "RATE_LIMIT_EXCEEDED":
            print("Rate limit exceeded after all retries")

            # Check rate limit info
            rate_limit_info = client.get_rate_limit_info()
            if rate_limit_info:
                from datetime import datetime
                reset_time = datetime.fromtimestamp(rate_limit_info.reset)
                print(f"Rate limit resets at: {reset_time}")

execute_with_retry_handling()

Monitoramento de uso

Acompanhe o uso e os limites da sua conta:

import os
from studio import StudioClient

client = StudioClient(
    api_key=os.getenv("STUDIO_API_KEY"),
    base_url="https://agent-studio.seeyu.ai"
)

def check_usage():
    try:
        limits = client.get_usage_limits()

        print("=== Rate Limits ===")
        print("Sync requests:")
        print(f"  Limit: {limits.rate_limit['sync']['limit']}")
        print(f"  Remaining: {limits.rate_limit['sync']['remaining']}")
        print(f"  Resets at: {limits.rate_limit['sync']['resetAt']}")
        print(f"  Is limited: {limits.rate_limit['sync']['isLimited']}")

        print("\nAsync requests:")
        print(f"  Limit: {limits.rate_limit['async']['limit']}")
        print(f"  Remaining: {limits.rate_limit['async']['remaining']}")
        print(f"  Resets at: {limits.rate_limit['async']['resetAt']}")
        print(f"  Is limited: {limits.rate_limit['async']['isLimited']}")

        print("\n=== Usage ===")
        print(f"Current period cost: ${limits.usage['currentPeriodCost']:.2f}")
        print(f"Limit: ${limits.usage['limit']:.2f}")
        print(f"Plan: {limits.usage['plan']}")

        percent_used = (limits.usage['currentPeriodCost'] / limits.usage['limit']) * 100
        print(f"Usage: {percent_used:.1f}%")

        if percent_used > 80:
            print("⚠️  Warning: You are approaching your usage limit!")

    except Exception as error:
        print(f"Error checking usage: {error}")

check_usage()

Execução de workflow com streaming

Execute workflows com respostas em streaming em tempo real:

from studio import StudioClient
import os

client = StudioClient(
    api_key=os.getenv("STUDIO_API_KEY"),
    base_url="https://agent-studio.seeyu.ai"
)

def execute_with_streaming():
    """Execute workflow with streaming enabled."""
    try:
        # Enable streaming for specific block outputs
        result = client.execute_workflow(
            "workflow-id",
            input={"message": "Count to five"},
            stream=True,
            selected_outputs=["agent1.content"]  # Use blockName.attribute format
        )

        print("Workflow result:", result)
    except Exception as error:
        print("Error:", error)

execute_with_streaming()

A resposta em streaming segue o formato Server-Sent Events (SSE):

data: {"blockId":"7b7735b9-19e5-4bd6-818b-46aae2596e9f","chunk":"One"}

data: {"blockId":"7b7735b9-19e5-4bd6-818b-46aae2596e9f","chunk":", two"}

data: {"event":"done","success":true,"output":{},"metadata":{"duration":610}}

data: [DONE]

Exemplo de streaming com Flask:

from flask import Flask, Response, stream_with_context
import requests
import json
import os

app = Flask(__name__)

@app.route('/stream-workflow')
def stream_workflow():
    """Stream workflow execution to the client."""

    def generate():
        response = requests.post(
            'https://agent-studio.seeyu.ai/api/v2/workflows/WORKFLOW_ID/execute',
            headers={
                'Content-Type': 'application/json',
                'X-API-Key': os.getenv('STUDIO_API_KEY')
            },
            json={
                'input': {'message': 'Generate a story'},
                'stream': True,
                'selectedOutputs': ['agent1.content']
            },
            stream=True
        )

        for line in response.iter_lines():
            if line:
                decoded_line = line.decode('utf-8')
                if decoded_line.startswith('data: '):
                    data = decoded_line[6:]  # Remove 'data: ' prefix

                    if data == '[DONE]':
                        break

                    try:
                        parsed = json.loads(data)
                        if 'chunk' in parsed:
                            yield f"data: {json.dumps(parsed)}\n\n"
                        elif parsed.get('event') == 'done':
                            yield f"data: {json.dumps(parsed)}\n\n"
                            print("Execution complete:", parsed.get('metadata'))
                    except json.JSONDecodeError:
                        pass

    return Response(
        stream_with_context(generate()),
        mimetype='text/event-stream'
    )

if __name__ == '__main__':
    app.run(debug=True)

Configuração por ambiente

Configure o cliente usando variáveis de ambiente:

import os
from studio import StudioClient

# Development configuration
client = StudioClient(
    api_key=os.getenv("STUDIO_API_KEY"),
    base_url=os.getenv("STUDIO_BASE_URL", "https://agent-studio.seeyu.ai")
)

import os
from studio import StudioClient

# Production configuration with error handling
api_key = os.getenv("STUDIO_API_KEY")
if not api_key:
    raise ValueError("STUDIO_API_KEY environment variable is required")

client = StudioClient(
    api_key=api_key,
    base_url=os.getenv("STUDIO_BASE_URL", "https://agent-studio.seeyu.ai")
)

Como obter sua chave de API

Acesse o Studio e entre na sua conta.

Vá até o workflow que você quer executar de forma programática.

Clique em "Deploy" para fazer o deploy do workflow, caso ele ainda não tenha deploy.

Durante o processo de deploy, selecione ou crie uma chave de API.

Copie a chave de API para usar na sua aplicação Python.

Requisitos

Python 3.8+
requests >= 2.25.0

Licença

Apache-2.0

Common Questions

Sim. Os workflows precisam ter deploy antes de serem executados pelo SDK. Use o método validate_workflow() para verificar se um workflow tem deploy e está pronto. Se ele retornar False, faça primeiro o deploy do workflow pela interface do Studio e crie ou selecione uma chave de API durante o deploy.

A execução síncrona (o padrão) bloqueia até o workflow terminar e devolve o resultado completo. A execução assíncrona (async_execution=True) retorna imediatamente com um ID de execução e uma URL de status que você pode consultar com get_workflow_run(). Use o modo assíncrono em workflows longos para evitar timeouts de requisição. Os status de execução são queued, pending, running, paused, completed, failed e cancelled.

O SDK traz suporte embutido a limites de requisições pelo método execute_with_retry(). Ele usa backoff exponencial (1s, 2s, 4s, 8s...) com 25% de jitter para evitar o efeito manada. Se a API retornar o header retry-after, esse valor é usado no lugar do cálculo. Você pode configurar max_retries, initial_delay, max_delay e backoff_multiplier. Use get_rate_limit_info() para consultar o estado atual do seu limite.

Sim. O StudioClient suporta o protocolo de context manager do Python. Use-o com a instrução 'with' para fechar a sessão HTTP subjacente automaticamente quando você terminar — o que é especialmente útil em scripts que criam e descartam instâncias do cliente.

O SDK levanta StudioError com uma propriedade code para os erros específicos da API. Os códigos mais comuns são UNAUTHORIZED (chave de API inválida), TIMEOUT (a requisição excedeu o tempo limite), RATE_LIMIT_EXCEEDED (requisições em excesso), USAGE_LIMIT_EXCEEDED (limite de cobrança atingido) e EXECUTION_ERROR (o workflow falhou). Use o código do erro para escrever um tratamento direcionado e a sua lógica de recuperação.

Use o método get_usage_limits() para consultar seu uso atual. Ele retorna os detalhes dos limites síncrono e assíncrono (limite, quanto resta, quando o contador reinicia e se você está limitado no momento), além do custo do período atual, do limite de uso e do seu plano. Assim você acompanha o consumo e cria alertas antes de bater no limite.

Python

Common Questions

Nesta página