Code Examples

Production-ready examples, plug and play

Quick Start

Get started in 5 minutes

Streaming

Real-time streaming output

Agent Workflow

Multi-step task execution

Guardrails

PII filtering & moderation

RAG Integration

Retrieval-augmented generation

Framework Integration

Next.js / Express / FastAPI

Quick Start

JavaScript / TypeScript

import { SkyAI } from "@skyaiapp/sdk";

// Initialize client
const sky = new SkyAI({
  apiKey: process.env.SKYAIAPP_API_KEY,
});

// Make a routing request
async function main() {
  const response = await sky.route({
    goal: "cost",
    strategy: "balanced",
    messages: [
      { role: "system", content: "You are a helpful assistant." },
      { role: "user", content: "What is the capital of France?" }
    ],
  });

  console.log(response.choices[0].message.content);
  console.log("Model used:", response.model);
  console.log("Cost:", response.routing.cost_usd);
}

main();

Python

import os
from skyaiapp import SkyAI

# Initialize client
sky = SkyAI(api_key=os.environ["SKYAIAPP_API_KEY"])

# Make a routing request
response = sky.route(
    goal="cost",
    strategy="balanced",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is the capital of France?"}
    ]
)

print(response.choices[0].message.content)
print(f"Model used: {response.model}")
print(f"Cost: ${response.routing.cost_usd}")

cURL

curl -X POST https://api.skyaiapp.com/v1/route \
  -H "Authorization: Bearer $SKYAIAPP_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "goal": "cost",
    "strategy": "balanced",
    "messages": [
      {"role": "system", "content": "You are a helpful assistant."},
      {"role": "user", "content": "What is the capital of France?"}
    ]
  }'

Streaming Response

TypeScript

const stream = await sky.route({
  goal: "quality",
  strategy: "quality-first",
  messages: [
    { role: "user", content: "Write a short story about AI..." }
  ],
  stream: true,
});

// Process stream chunks
for await (const chunk of stream) {
  if (chunk.choices[0]?.delta?.content) {
    process.stdout.write(chunk.choices[0].delta.content);
  }
}

// Get final usage stats
console.log("\n\nUsage:", stream.usage);

Python

stream = sky.route(
    goal="quality",
    strategy="quality-first",
    messages=[
        {"role": "user", "content": "Write a short story about AI..."}
    ],
    stream=True
)

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

print(f"\n\nUsage: {stream.usage}")

Agent Workflow

TypeScript - Full Agent Example

import { SkyAI } from "@skyaiapp/sdk";

const sky = new SkyAI({ apiKey: process.env.SKYAIAPP_API_KEY });

// Create an agent with tools
const agent = sky.createAgent({
  tools: ["web_search", "read_url", "calculator", "code_exec"],
  maxSteps: 15,
  timeout: 300000, // 5 minutes
  sandbox: true,   // Run code in isolated sandbox
});

// Define step callback for real-time monitoring
agent.onStep((step) => {
  console.log(`Step ${step.number}: ${step.action}`);
  console.log(`  Input: ${JSON.stringify(step.input)}`);
  console.log(`  Duration: ${step.duration_ms}ms`);
});

// Run the agent
async function runResearchTask() {
  const result = await agent.run({
    task: `
      Research the top 5 AI companies by market cap in 2024.
      For each company, find:
      1. Current market capitalization
      2. Main AI products/services
      3. Recent AI announcements
      
      Compile findings into a structured report.
    `,
  });

  console.log("\n=== Agent Result ===");
  console.log(result.output);
  console.log("\nTotal steps:", result.steps.length);
  console.log("Total cost:", result.usage.total_cost_usd);
}

runResearchTask();

Python - With Custom Tools

from skyaiapp import SkyAI, Tool

sky = SkyAI(api_key=os.environ["SKYAIAPP_API_KEY"])

# Define custom tool
@sky.tool("fetch_stock_price")
def fetch_stock_price(symbol: str) -> dict:
    """Fetch current stock price for a given symbol."""
    # Your implementation here
    return {"symbol": symbol, "price": 150.25, "currency": "USD"}

# Create agent with custom tool
agent = sky.create_agent(
    tools=["web_search", "calculator", "fetch_stock_price"],
    max_steps=10
)

# Run with callback
def on_step(step):
    print(f"[Step {step.number}] {step.action}: {step.duration_ms}ms")

result = agent.run(
    task="Get Apple stock price and calculate PE ratio if earnings are $6.42",
    on_step=on_step
)

print(result.output)

Guardrails

PII Filtering

const response = await sky.route({
  goal: "quality",
  messages: [
    { 
      role: "user", 
      content: "My email is john@example.com and SSN is 123-45-6789" 
    }
  ],
  guardrails: {
    pii_filter: {
      enabled: true,
      types: ["email", "ssn", "phone", "credit_card"],
      action: "redact", // or "block"
    }
  }
});

// Original PII is automatically redacted
// "My email is [EMAIL] and SSN is [SSN]"

Content Moderation

const response = await sky.route({
  goal: "quality",
  messages: [
    { role: "user", content: userInput }
  ],
  guardrails: {
    moderation: {
      enabled: true,
      categories: ["hate", "violence", "self_harm", "sexual"],
      threshold: 0.7,
      action: "block"
    },
    topic_filter: {
      enabled: true,
      blocked_topics: ["illegal_activities", "weapons"],
    }
  }
});

if (response.guardrails?.blocked) {
  console.log("Content blocked:", response.guardrails.reason);
}

RAG Integration

Retrieval-Augmented Generation

import { SkyAI } from "@skyaiapp/sdk";

const sky = new SkyAI({ apiKey: process.env.SKYAIAPP_API_KEY });

// 1. Create a knowledge base
const kb = await sky.knowledgeBase.create({
  name: "product-docs",
  description: "Product documentation and FAQs",
});

// 2. Ingest documents
await sky.knowledgeBase.ingest(kb.id, {
  documents: [
    { content: "...", metadata: { source: "docs/api.md" } },
    { content: "...", metadata: { source: "docs/faq.md" } },
  ],
  chunking: { strategy: "semantic", max_tokens: 512 },
});

// 3. Query with RAG
const response = await sky.route({
  goal: "quality",
  messages: [
    { role: "user", content: "How do I authenticate with the API?" }
  ],
  rag: {
    knowledge_base_id: kb.id,
    top_k: 5,
    min_relevance: 0.7,
    include_sources: true,
  }
});

console.log(response.choices[0].message.content);
console.log("Sources:", response.rag.sources);

Framework Integration

Next.js App Router

// app/api/chat/route.ts
import { SkyAI } from "@skyaiapp/sdk";
import { NextResponse } from "next/server";

const sky = new SkyAI({ apiKey: process.env.SKYAIAPP_API_KEY });

export async function POST(req: Request) {
  const { messages } = await req.json();

  const stream = await sky.route({
    goal: "balanced",
    messages,
    stream: true,
  });

  // Return streaming response
  return new Response(stream.toReadableStream(), {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
    },
  });
}

// Client component usage
"use client";
import { useChat } from "@skyaiapp/react";

export function ChatComponent() {
  const { messages, input, handleInputChange, handleSubmit, isLoading } = useChat({
    api: "/api/chat",
  });

  return (
    <form onSubmit={handleSubmit}>
      {messages.map((m) => <div key={m.id}>{m.content}</div>)}
      <input value={input} onChange={handleInputChange} />
      <button type="submit" disabled={isLoading}>Send</button>
    </form>
  );
}

Express.js

import express from "express";
import { SkyAI } from "@skyaiapp/sdk";

const app = express();
const sky = new SkyAI({ apiKey: process.env.SKYAIAPP_API_KEY });

app.use(express.json());

app.post("/api/chat", async (req, res) => {
  const { messages } = req.body;

  try {
    const response = await sky.route({
      goal: "cost",
      messages,
    });

    res.json(response);
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
});

// Streaming endpoint
app.post("/api/chat/stream", async (req, res) => {
  const { messages } = req.body;

  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");

  const stream = await sky.route({ goal: "quality", messages, stream: true });

  for await (const chunk of stream) {
    res.write(`data: ${JSON.stringify(chunk)}\n\n`);
  }

  res.end();
});

app.listen(3000);

FastAPI (Python)

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from skyaiapp import SkyAI
import os

app = FastAPI()
sky = SkyAI(api_key=os.environ["SKYAIAPP_API_KEY"])

class ChatRequest(BaseModel):
    messages: list[dict]

@app.post("/api/chat")
async def chat(request: ChatRequest):
    response = sky.route(
        goal="cost",
        messages=request.messages
    )
    return response.model_dump()

@app.post("/api/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate():
        stream = sky.route(
            goal="quality",
            messages=request.messages,
            stream=True
        )
        for chunk in stream:
            yield f"data: {chunk.model_dump_json()}\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )

Need more examples?

Check out our GitHub for full example projects

GitHub Examples API Docs

Error Handling

Streaming

Was this page helpful?

Let us know how we can improve