synova-cloud-sdk 1.1.0

Official Python SDK for Synova Cloud API

Synova Cloud SDK for Python

Official Python SDK for the Synova Cloud API.

Sync and async clients, full prompt execution (incl. streaming + agentic loops), LLM model catalog, file uploads, and observability (traces / spans).

Installation

pip install synova-cloud-sdk

Requires Python 3.10+.

Quick start

from synova_cloud_sdk import SynovaCloudSDK

client = SynovaCloudSDK("your-api-key")

response = client.prompts.execute(
    "prm_abc123",
    provider="openai",
    model="gpt-4o",
    variables={"name": "World"},
)
print(response.content)

Async equivalent:

import asyncio
from synova_cloud_sdk import AsyncSynovaCloudSDK

async def main():
    async with AsyncSynovaCloudSDK("your-api-key") as client:
        response = await client.prompts.execute(
            "prm_abc123",
            provider="openai",
            model="gpt-4o",
            variables={"name": "World"},
        )
        print(response.content)

asyncio.run(main())

Configuration

from synova_cloud_sdk import SynovaCloudSDK, RetryConfig

client = SynovaCloudSDK(
    "your-api-key",
    base_url="https://api.synova.cloud",     # optional override
    timeout_seconds=30.0,
    debug=True,
    retry=RetryConfig(
        max_retries=3,
        strategy="exponential",       # or "linear"
        initial_delay_seconds=1.0,
        max_delay_seconds=30.0,
        backoff_multiplier=2.0,
    ),
)

Naming. The Python SDK uses snake_case for all attributes and keyword arguments. The wire format is camelCase — handled automatically by pydantic aliases. So response.trace_id and response.execution_usage.input_tokens work as expected without any manual conversion.

Timeouts. All durations are seconds (not milliseconds — that's the Node SDK convention).

Prompts

Get a prompt

client.prompts.get("prm_abc123")                      # latest
client.prompts.get("prm_abc123", tag="production")
client.prompts.get("prm_abc123", version="1.2.0")

Execute

response = client.prompts.execute(
    "prm_abc123",
    provider="openai",            # or "anthropic", "google", "azure_openai", ...
    model="gpt-4o",
    variables={"topic": "AI"},
    parameters={"temperature": 0.7, "maxTokens": 1000},
)
print(response.content)
print(response.execution_usage.total_tokens)

With your own provider key

client.prompts.execute(
    "prm_abc123",
    provider="anthropic",
    model="claude-sonnet-4-20250514",
    api_key="sk-ant-...",
)

# Azure OpenAI requires endpoint
client.prompts.execute(
    "prm_abc123",
    provider="azure_openai",
    model="my-deployment",
    api_key="...",
    azure_endpoint="https://my-resource.openai.azure.com",
)

Multi-turn conversation

client.prompts.execute(
    "prm_chat",
    provider="anthropic",
    model="claude-sonnet-4-20250514",
    messages=[
        {"role": "user", "content": "What is TypeScript?"},
        {"role": "assistant", "content": "TypeScript is..."},
        {"role": "user", "content": "How do I use generics?"},
    ],
)

Tag / version variants

client.prompts.execute_by_tag("prm_abc123", "production", provider="openai", model="gpt-4o")
client.prompts.execute_by_version("prm_abc123", "1.2.0", provider="openai", model="gpt-4o")

Tools (function calling)

If a tool has an execute handler, the SDK runs an agentic loop automatically.

from pydantic import BaseModel
from synova_cloud_sdk import Tool

class WeatherArgs(BaseModel):
    city: str

def get_weather(args: dict) -> dict:
    return {"temperature": 21, "condition": "sunny"}

response = client.prompts.execute(
    "prm_agent",
    provider="anthropic",
    model="claude-sonnet-4-20250514",
    variables={"question": "Weather in Paris?"},
    tools={
        "get_weather": Tool(
            description="Get current weather for a city",
            parameters=WeatherArgs,         # pydantic class — auto-converted to JSON Schema
            execute=get_weather,
        ),
    },
    max_tool_roundtrips=10,                 # default 25, hard cap 100
    tool_timeout_seconds=60.0,
)

In the async client, the same handler may be async def:

async def get_weather(args: dict) -> dict:
    return await fetch_weather_async(args["city"])

Tools without execute return tool_calls for manual handling:

response = client.prompts.execute(
    "prm_abc123",
    provider="openai",
    model="gpt-4o",
    tools={"get_weather": Tool(description="Get weather", parameters=WeatherArgs)},
)
if response.type == "tool_calls":
    for call in response.tool_calls:
        print(call.name, call.arguments)

Structured output

Pass a pydantic model class as response_schema to get a typed .object:

from pydantic import BaseModel
from synova_cloud_sdk import SynovaCloudSDK

class Article(BaseModel):
    title: str
    keywords: list[str]
    priority: int

client = SynovaCloudSDK("your-api-key")
response = client.prompts.execute(
    "prm_abc123",
    provider="openai",
    model="gpt-4o",
    response_schema=Article,
)
article: Article = response.object
print(article.title, article.keywords)

Or pass a raw JSON Schema dict — .object will be a regular dict:

response = client.prompts.execute(
    "prm_abc123",
    provider="openai",
    model="gpt-4o",
    response_schema={
        "type": "object",
        "properties": {
            "title": {"type": "string"},
            "keywords": {"type": "array", "items": {"type": "string"}},
        },
        "required": ["title", "keywords"],
    },
)

Streaming

for event in client.prompts.stream(
    "prm_abc123",
    provider="anthropic",
    model="claude-sonnet-4-20250514",
):
    if event.type == "text-delta":
        print(event.text_delta, end="", flush=True)
    elif event.type == "finish":
        print("\nUsage:", event.usage.total_tokens)

Async:

async for event in client.prompts.stream(
    "prm_abc123",
    provider="anthropic",
    model="claude-sonnet-4-20250514",
):
    if event.type == "text-delta":
        print(event.text_delta, end="", flush=True)

Streaming agentic loop

If any tool has execute, stream() interleaves tool execution with output. You'll see tool-call → roundtrip-finish → tool-result → next round → final finish.

for event in client.prompts.stream(
    "prm_abc123",
    provider="anthropic",
    model="claude-sonnet-4-20250514",
    tools={"get_weather": Tool(description="...", parameters=WeatherArgs, execute=get_weather)},
):
    match event.type:
        case "text-delta":
            print(event.text_delta, end="", flush=True)
        case "tool-call":
            print(f"\n→ calling {event.tool_name}({event.args})")
        case "roundtrip-finish":
            print(f"\n[roundtrip {event.roundtrip} done]")
        case "tool-result":
            print(f"← {event.tool_call_id}: {event.result}")
        case "finish":
            print("\nDone.")

Image generation

response = client.prompts.generate_image(
    "prm_image",
    provider="openai",
    model="dall-e-3",
    variables={"description": "A sunset over mountains"},
)
for f in response.files or []:
    print(f.url, f.mime_type)

Models

result = client.models.list()
for provider in result.providers:
    print(provider.display_name)
    for model in provider.models:
        print(f"  - {model.display_name} ({model.id})")

# Filtering
client.models.list(type="image")
client.models.list(capability="vision")
client.models.list(provider="openai")

# By provider
models = client.models.get_by_provider("anthropic")

# Single model
model = client.models.get("openai", "gpt-4o")
print(model.limits.context_window)

Files

from pathlib import Path

# bytes, Path, or (name, bytes, mime) tuples — all accepted
result = client.files.upload(
    [Path("photo.png"), b"binary data", ("doc.pdf", b"%PDF-1.4", "application/pdf")],
    project_id="prj_abc123",
)
file_id = result.data[0].id

# Use in a vision call
client.prompts.execute(
    "prm_vision",
    provider="openai",
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "What's in this image?", "files": [{"fileId": file_id}]},
    ],
)

Observability — traces and spans

Every execute() automatically creates a trace + span. The returned response exposes trace_id and span_id. Pass trace_id to subsequent calls to group them under the same trace.

r1 = client.prompts.execute("prm_x", provider="openai", model="gpt-4o")
r2 = client.prompts.execute(
    "prm_y",
    provider="openai",
    model="gpt-4o",
    trace_id=r1.trace_id,
)

Custom spans

trace = client.traces.create(project_id="prj_abc123", name="checkout-flow")

# Wrapper — auto end + error reporting
docs = client.spans.wrap(
    trace_id=trace.id,
    type="retriever",
    name="vector_search",
    input={"query": "...", "top_k": 5},
    fn=lambda: vector_db.search("..."),
)

weather = client.spans.wrap_tool(
    trace_id=trace.id,
    tool_name="fetch_weather",
    args={"city": "NYC"},
    fn=lambda args: fetch_weather(args["city"]),
)

# Manual
span = client.spans.create(trace.id, type="custom", name="step")
try:
    result = do_work()
    client.spans.end(span.id, status="completed", output=result)
except Exception as exc:
    client.spans.end(span.id, status="error", status_message=str(exc))
    raise

Errors

from synova_cloud_sdk import (
    ApiSynovaError,
    AuthSynovaError,
    NotFoundSynovaError,
    RateLimitSynovaError,
    ServerSynovaError,
    TimeoutSynovaError,
    NetworkSynovaError,
    ExecutionSynovaError,
    SynovaError,
)

try:
    client.prompts.execute("prm_abc123", provider="openai", model="gpt-4o")
except AuthSynovaError:
    ...                                # 401 — bad API key
except NotFoundSynovaError:
    ...                                # 404 — prompt missing
except RateLimitSynovaError as exc:
    print("retry after", exc.retry_after_seconds)
except ServerSynovaError:
    ...                                # 5xx
except TimeoutSynovaError as exc:
    print(f"timed out after {exc.timeout_seconds}s")
except NetworkSynovaError:
    ...                                # transport failure
except ExecutionSynovaError as exc:
    print(f"LLM error [{exc.code}]: {exc.message}")
except ApiSynovaError as exc:
    print(f"API error [{exc.code}] req={exc.request_id}")
except SynovaError:
    ...                                # catch-all

Retry behavior

• Retried automatically: 429 (Retry-After honored), 5xx, network errors, timeouts.
• Not retried: 401, 404, other 4xx.

Schema converters

Use any registered adapter directly:

from synova_cloud_sdk import SchemaResolver

# Pydantic class → JSON Schema dict
SchemaResolver.resolve(MyModel)

# Raw dict → cleaned (nullable transforms applied)
SchemaResolver.resolve({"type": "object", "properties": {...}})

Custom adapter:

from synova_cloud_sdk import SchemaResolver
from synova_cloud_sdk.schema.adapters import SchemaAdapter

class MyAdapter:
    def can_handle(self, schema): ...
    def to_json_schema(self, schema): ...

SchemaResolver.register_adapter(MyAdapter(), position="first")

License

MIT