pip install requests openai

import os

API_KEY = os.environ.get("NOVAKIT_API_KEY")
BASE_URL = "https://www.novakit.ai/api/v1"

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

from openai import OpenAI

client = OpenAI(
    api_key=os.environ.get("NOVAKIT_API_KEY"),
    base_url="https://www.novakit.ai/api/v1"
)

stream = client.chat.completions.create(
    model="openai/gpt-4o-mini",
    messages=[{"role": "user", "content": "Write a short poem"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

import requests

response = requests.post(
    f"{BASE_URL}/agents",
    headers=headers,
    json={
        "name": "Research Assistant",
        "description": "Helps with research and information gathering",
        "system_prompt": "You are a helpful research assistant. Search the web to find accurate information and cite your sources.",
        "model": "anthropic/claude-3.5-sonnet",
        "tools": ["web_search", "web_fetch", "create_document"],
        "max_iterations": 10,
        "memory_enabled": True
    }
)

agent = response.json()["agent"]
print(f"Created agent: {agent['id']}")

import requests
import json

def run_agent(agent_id, user_input, session_id=None):
    """Run an agent and stream the response."""
    payload = {"input": user_input}
    if session_id:
        payload["session_id"] = session_id

    response = requests.post(
        f"{BASE_URL}/agents/{agent_id}/run",
        headers=headers,
        json=payload,
        stream=True
    )

    result = None
    for line in response.iter_lines():
        if not line:
            continue

        line = line.decode('utf-8')
        if not line.startswith('data: '):
            continue

        event = json.loads(line[6:])
        event_type = event.get("type")

        if event_type == "start":
            print(f"Run started: {event['run_id']}")

        elif event_type == "step":
            step = event.get("step", {})
            step_type = step.get("type")

            if step_type == "thinking":
                print(f"[Thinking] {step.get('content', '')[:100]}...")
            elif step_type == "tool_call":
                print(f"[Tool] Calling {step.get('tool')}...")
            elif step_type == "tool_result":
                print(f"[Result] Got response from {step.get('tool')}")
            elif step_type == "response":
                print(f"\n[Response]\n{step.get('content', '')}")

        elif event_type == "done":
            result = event
            print(f"\nCompleted! Credits used: {event.get('credits_used')}")

        elif event_type == "error":
            print(f"Error: {event.get('error')}")

    return result

# Usage
result = run_agent(
    agent_id="agent_abc123",
    user_input="Research the latest developments in quantum computing"
)

import uuid

# Create a session for conversation continuity
session_id = str(uuid.uuid4())

# First turn
run_agent(agent["id"], "Search for AI news from today", session_id=session_id)

# Second turn - references previous context
run_agent(agent["id"], "Tell me more about the first result", session_id=session_id)

# Third turn
run_agent(agent["id"], "Create a summary document of what we discussed", session_id=session_id)

import requests

# Upload a document for RAG
with open("company_docs.pdf", "rb") as f:
    response = requests.post(
        f"{BASE_URL}/agents/{agent_id}/knowledge",
        headers={"Authorization": f"Bearer {API_KEY}"},
        files={"file": f}
    )

file_info = response.json()["file"]
print(f"Uploaded: {file_info['id']}, Status: {file_info['status']}")

# Wait for processing
import time
while True:
    status = requests.get(
        f"{BASE_URL}/agents/{agent_id}/knowledge/{file_info['id']}",
        headers=headers
    ).json()

    if status["file"]["status"] == "completed":
        print("Knowledge file ready!")
        break
    elif status["file"]["status"] == "failed":
        print(f"Failed: {status['file'].get('error_message')}")
        break

    time.sleep(2)

response = requests.get(
    f"{BASE_URL}/agents/{agent_id}/runs?limit=10",
    headers=headers
)

for run in response.json()["runs"]:
    print(f"{run['id']}: {run['status']} - {run['input'][:50]}...")

import requests

response = requests.post(
    f"{BASE_URL}/images/generations",
    headers=headers,
    json={
        "prompt": "A beautiful sunset over the ocean",
        "model": "fal-ai/flux/dev",
        "size": "1024x1024"
    }
)

data = response.json()
image_url = data["data"][0]["url"]
print(f"Image URL: {image_url}")

import requests
import time

# Start async job
response = requests.post(
    f"{BASE_URL}/videos/generations?async=true",
    headers=headers,
    json={
        "prompt": "A timelapse of clouds moving over mountains",
        "duration_seconds": 5
    }
)

job_id = response.json()["id"]
print(f"Job started: {job_id}")

# Poll for completion
while True:
    status = requests.get(
        f"{BASE_URL}/jobs/{job_id}?poll=true",
        headers=headers
    ).json()

    print(f"Status: {status['status']}, Progress: {status.get('progress', 0)}%")

    if status["status"] == "completed":
        video_url = status["outputData"]["video"]["url"]
        print(f"Video URL: {video_url}")
        break
    elif status["status"] == "failed":
        print(f"Failed: {status.get('errorMessage')}")
        break

    time.sleep(5)

import requests

response = requests.post(
    f"{BASE_URL}/audio/speech",
    headers=headers,
    json={
        "input": "Hello, welcome to NovaKit!",
        "voice": "nova",
        "speed": 1.0
    }
)

audio_url = response.json()["audio_url"]

# Download the audio file
audio_response = requests.get(audio_url)
with open("output.mp3", "wb") as f:
    f.write(audio_response.content)

import requests

response = requests.post(
    f"{BASE_URL}/audio/transcriptions",
    headers=headers,
    json={
        "file": "https://example.com/audio.mp3",
        "language": "auto",
        "diarization": True
    }
)

data = response.json()
print(f"Transcript: {data['text']}")
print(f"Duration: {data['duration']} seconds")

# Print speaker segments
for segment in data.get("segments", []):
    speaker = segment.get("speaker", "Unknown")
    print(f"[{speaker}] {segment['text']}")

import requests

def make_request(endpoint, data):
    response = requests.post(
        f"{BASE_URL}/{endpoint}",
        headers=headers,
        json=data
    )

    if response.status_code == 200:
        return response.json()

    error = response.json()

    if response.status_code == 401:
        raise Exception(f"Authentication failed: {error['error']}")
    elif response.status_code == 402:
        raise Exception(f"Quota exceeded: {error['error']}")
    elif response.status_code == 429:
        retry_after = error.get("retry_after", 60)
        raise Exception(f"Rate limited. Retry after {retry_after}s")
    else:
        raise Exception(f"API error: {error['error']}")

import requests
import json
import os

class NovaKitClient:
    def __init__(self, api_key=None):
        self.api_key = api_key or os.environ.get("NOVAKIT_API_KEY")
        self.base_url = "https://www.novakit.ai/api/v1"
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

    def chat(self, messages, model="openai/gpt-4o-mini", **kwargs):
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=self.headers,
            json={"messages": messages, "model": model, **kwargs}
        )
        response.raise_for_status()
        return response.json()

    def generate_image(self, prompt, **kwargs):
        response = requests.post(
            f"{self.base_url}/images/generations",
            headers=self.headers,
            json={"prompt": prompt, **kwargs}
        )
        response.raise_for_status()
        return response.json()

    def create_agent(self, name, system_prompt, tools=None, **kwargs):
        response = requests.post(
            f"{self.base_url}/agents",
            headers=self.headers,
            json={
                "name": name,
                "system_prompt": system_prompt,
                "tools": tools or [],
                **kwargs
            }
        )
        response.raise_for_status()
        return response.json()["agent"]

    def run_agent(self, agent_id, input_text, session_id=None):
        payload = {"input": input_text}
        if session_id:
            payload["session_id"] = session_id

        response = requests.post(
            f"{self.base_url}/agents/{agent_id}/run",
            headers=self.headers,
            json=payload,
            stream=True
        )

        events = []
        for line in response.iter_lines():
            if line and line.startswith(b'data: '):
                events.append(json.loads(line[6:]))
        return events

# Usage
client = NovaKitClient()

# Chat
response = client.chat([{"role": "user", "content": "Hello!"}])
print(response["choices"][0]["message"]["content"])

# Create and run agent
agent = client.create_agent(
    name="Helper",
    system_prompt="You are helpful.",
    tools=["web_search"]
)
events = client.run_agent(agent["id"], "What's the weather in NYC?")