Endpoints

Overview

@Endpoint() turns Python functions into serverless API endpoints - like AWS Lambda, but simpler.

from ziet import Endpoint, db

@Endpoint(
    id="get_user",
    name="Get User",
    description="Fetch user by ID",
    method="GET",
    path="/users/{user_id}"
)
def get_user(user_id: str) -> dict:
    """Fetch user from database."""
    
    user = db.query("users").where("id", user_id).first()
    
    return {
        "status": 200,
        "body": user
    }

Deploy once, invoke anywhere:

# Deploy
ziet deploy

# Invoke
curl https://api.ziet.ai/endpoints/get_user/abc123

What are Endpoints?

Endpoints are serverless API functions that:

Respond to HTTP requests (GET, POST, PUT, DELETE)
Scale automatically with traffic
Return JSON responses
Have built-in authentication
Support path parameters and query strings

When to use Endpoints vs Agents:

Feature	Endpoints	Agents
Use case	API endpoints, webhooks	Complex workflows, multi-step tasks
Execution	Single function call	Multiple actions, orchestration
Response time	< 1s (fast)	Seconds to minutes
Strategies	❌ No	✅ Yes
Best for	REST APIs, webhooks	Research, booking, automation

Building Endpoints

Basic Endpoint

from ziet import Endpoint

@Endpoint(
    id="hello",
    name="Hello World",
    description="Simple hello endpoint",
    method="GET",
    path="/hello"
)
def hello() -> dict:
    """Return hello message."""
    return {
        "status": 200,
        "body": {"message": "Hello, world!"}
    }

Invoke:

curl https://api.ziet.ai/endpoints/hello

Endpoint Parameters

string

required

Unique endpoint identifier (alphanumeric, underscores, hyphens)

name

string

required

Human-readable name

description

string

required

What the endpoint does

method

string

default:"POST"

HTTP method: GET, POST, PUT, DELETE, PATCH

path

string

Custom URL path (e.g., /users/{id})If not provided, defaults to /endpoints/{id}

auth

boolean

default:"true"

Require API key authentication

timeout

int

default:"30"

Max execution time in seconds

rate_limit

int

Max requests per minute

HTTP Methods

GET Endpoint

Fetch data:

from ziet import Endpoint, db

@Endpoint(
    id="list_products",
    name="List Products",
    description="Get all products",
    method="GET",
    path="/products"
)
def list_products(category: str = None, limit: int = 10) -> dict:
    """List products with optional filtering."""
    query = db.query("products")
    
    if category:
        query = query.where("category", category)
    
    products = query.limit(limit).all()
    
    return {
        "status": 200,
        "body": {
            "products": products,
            "count": len(products)
        }
    }

Invoke:

curl "https://api.ziet.ai/products?category=electronics&limit=5"

POST Endpoint

Create data:

from ziet import Endpoint, db

@Endpoint(
    id="create_user",
    name="Create User",
    description="Create a new user",
    method="POST",
    path="/users"
)
def create_user(email: str, name: str, role: str = "user") -> dict:
    """Create a new user."""
    # Insert into database
    user_id = db.insert("users", {
        "email": email,
        "name": name,
        "role": role,
        "created_at": "now()"
    })
    
    return {
        "status": 201,
        "body": {
            "id": user_id,
            "email": email,
            "name": name,
            "role": role
        }
    }

Invoke:

curl -X POST https://api.ziet.ai/users \
  -H "Content-Type: application/json" \
  -d '{"email": "alice@example.com", "name": "Alice"}'

PUT Endpoint

Update data:

from ziet import Endpoint, db

@Endpoint(
    id="update_user",
    name="Update User",
    description="Update user details",
    method="PUT",
    path="/users/{user_id}"
)
def update_user(user_id: str, name: str = None, role: str = None) -> dict:
    """Update user."""
    updates = {}
    if name:
        updates["name"] = name
    if role:
        updates["role"] = role
    
    db.update("users", updates).where("id", user_id).execute()
    
    return {
        "status": 200,
        "body": {"id": user_id, "updated": True}
    }

Invoke:

curl -X PUT https://api.ziet.ai/users/abc123 \
  -H "Content-Type: application/json" \
  -d '{"name": "Alice Smith", "role": "admin"}'

DELETE Endpoint

Delete data:

from ziet import Endpoint, db

@Endpoint(
    id="delete_user",
    name="Delete User",
    description="Delete a user",
    method="DELETE",
    path="/users/{user_id}"
)
def delete_user(user_id: str) -> dict:
    """Delete user."""
    db.delete("users").where("id", user_id).execute()
    
    return {
        "status": 200,
        "body": {"id": user_id, "deleted": True}
    }

Invoke:

curl -X DELETE https://api.ziet.ai/users/abc123

Path Parameters

Use {parameter} in path:

from ziet import Endpoint, db

@Endpoint(
    id="get_post",
    name="Get Post",
    description="Get blog post by ID",
    method="GET",
    path="/posts/{post_id}"
)
def get_post(post_id: str) -> dict:
    """Fetch post by ID."""
    post = db.query("posts").where("id", post_id).first()
    
    if not post:
        return {
            "status": 404,
            "body": {"error": "Post not found"}
        }
    
    return {
        "status": 200,
        "body": post
    }

Invoke:

curl https://api.ziet.ai/posts/abc123

Query Parameters

Accept query strings as function parameters:

from ziet import Endpoint, db

@Endpoint(
    id="search_posts",
    name="Search Posts",
    description="Search blog posts",
    method="GET",
    path="/posts/search"
)
def search_posts(q: str, limit: int = 10, offset: int = 0) -> dict:
    """Search posts by query."""
    posts = db.query("posts") \
        .where("title", "LIKE", f"%{q}%") \
        .limit(limit) \
        .offset(offset) \
        .all()
    
    return {
        "status": 200,
        "body": {
            "results": posts,
            "count": len(posts),
            "query": q
        }
    }

Invoke:

curl "https://api.ziet.ai/posts/search?q=python&limit=5&offset=0"

Response Format

Return a dict with status and body:

@Endpoint(id="example", method="GET", path="/example")
def example() -> dict:
    return {
        "status": 200,              # HTTP status code
        "body": {"key": "value"},   # Response body (JSON)
        "headers": {                # Optional: custom headers
            "X-Custom": "value"
        }
    }

Status codes:

200 - OK
201 - Created
400 - Bad Request
401 - Unauthorized
404 - Not Found
500 - Internal Server Error

Authentication

Require Auth (Default)

By default, endpoints require API key authentication:

@Endpoint(
    id="protected",
    method="GET",
    path="/protected"
)
def protected() -> dict:
    return {"status": 200, "body": {"message": "Authenticated!"}}

Invoke with API key:

curl https://api.ziet.ai/protected \
  -H "Authorization: Bearer YOUR_API_KEY"

Public Endpoint

Disable authentication for public endpoints:

@Endpoint(
    id="public",
    method="GET",
    path="/public",
    auth=False  # No API key required
)
def public() -> dict:
    return {"status": 200, "body": {"message": "Public endpoint"}}

Invoke without auth:

curl https://api.ziet.ai/public

Rate Limiting

Limit requests per minute:

@Endpoint(
    id="limited",
    method="GET",
    path="/limited",
    rate_limit=10  # Max 10 requests per minute
)
def limited() -> dict:
    return {"status": 200, "body": {"message": "Rate limited"}}

Using Database

Store and query data:

from ziet import Endpoint, db

@Endpoint(
    id="create_order",
    name="Create Order",
    description="Create new order",
    method="POST",
    path="/orders"
)
def create_order(user_id: str, items: list, total: float) -> dict:
    """Create order and store in database."""
    # Insert order
    order_id = db.insert("orders", {
        "user_id": user_id,
        "total": total,
        "status": "pending",
        "created_at": "now()"
    })
    
    # Insert order items
    for item in items:
        db.insert("order_items", {
            "order_id": order_id,
            "product_id": item["product_id"],
            "quantity": item["quantity"],
            "price": item["price"]
        })
    
    return {
        "status": 201,
        "body": {
            "order_id": order_id,
            "total": total,
            "items": len(items)
        }
    }

Using Integrations

Call external APIs:

from ziet import Endpoint
from ziet.integrations import stripe, sendgrid

@Endpoint(
    id="checkout",
    name="Checkout",
    description="Process checkout and send confirmation",
    method="POST",
    path="/checkout"
)
def checkout(email: str, amount: int, items: list) -> dict:
    """Process payment and send confirmation."""
    # Charge with Stripe
    payment = stripe.create_payment_intent(
        amount=amount * 100,
        currency="usd"
    )
    
    # Send confirmation email
    sendgrid.send(
        to=email,
        subject="Order Confirmed",
        body=f"<h1>Order Confirmed</h1><p>Total: ${amount}</p>"
    )
    
    return {
        "status": 200,
        "body": {
            "payment_id": payment["id"],
            "amount": amount,
            "email": email
        }
    }

Complete Example: REST API

from ziet import Endpoint, db

# List all todos
@Endpoint(
    id="list_todos",
    method="GET",
    path="/todos"
)
def list_todos(status: str = None) -> dict:
    """List todos with optional status filter."""
    query = db.query("todos")
    
    if status:
        query = query.where("status", status)
    
    todos = query.all()
    
    return {
        "status": 200,
        "body": {"todos": todos}
    }

# Create todo
@Endpoint(
    id="create_todo",
    method="POST",
    path="/todos"
)
def create_todo(title: str, description: str = "") -> dict:
    """Create a new todo."""
    todo_id = db.insert("todos", {
        "title": title,
        "description": description,
        "status": "pending",
        "created_at": "now()"
    })
    
    return {
        "status": 201,
        "body": {"id": todo_id, "title": title}
    }

# Get single todo
@Endpoint(
    id="get_todo",
    method="GET",
    path="/todos/{todo_id}"
)
def get_todo(todo_id: str) -> dict:
    """Get todo by ID."""
    todo = db.query("todos").where("id", todo_id).first()
    
    if not todo:
        return {"status": 404, "body": {"error": "Todo not found"}}
    
    return {"status": 200, "body": todo}

# Update todo
@Endpoint(
    id="update_todo",
    method="PUT",
    path="/todos/{todo_id}"
)
def update_todo(todo_id: str, status: str = None, title: str = None) -> dict:
    """Update todo."""
    updates = {}
    if status:
        updates["status"] = status
    if title:
        updates["title"] = title
    
    db.update("todos", updates).where("id", todo_id).execute()
    
    return {"status": 200, "body": {"id": todo_id, "updated": True}}

# Delete todo
@Endpoint(
    id="delete_todo",
    method="DELETE",
    path="/todos/{todo_id}"
)
def delete_todo(todo_id: str) -> dict:
    """Delete todo."""
    db.delete("todos").where("id", todo_id).execute()
    
    return {"status": 200, "body": {"id": todo_id, "deleted": True}}

Deploy:

ziet deploy

Use your API:

# List todos
curl https://api.ziet.ai/todos

# Create todo
curl -X POST https://api.ziet.ai/todos \
  -d '{"title": "Buy groceries"}'

# Get todo
curl https://api.ziet.ai/todos/abc123

# Update todo
curl -X PUT https://api.ziet.ai/todos/abc123 \
  -d '{"status": "completed"}'

# Delete todo
curl -X DELETE https://api.ziet.ai/todos/abc123

Best Practices

Use appropriate HTTP methods

Follow REST conventions:

GET - Fetch data (read-only)
POST - Create resources
PUT - Update resources (full update)
PATCH - Partial update
DELETE - Delete resources

Return proper status codes

# Success
return {"status": 200, "body": {...}}

# Created
return {"status": 201, "body": {...}}

# Not found
return {"status": 404, "body": {"error": "Not found"}}

# Bad request
return {"status": 400, "body": {"error": "Invalid input"}}

Handle errors gracefully

@Endpoint(id="safe", method="GET", path="/safe")
def safe(user_id: str) -> dict:
    try:
        user = db.query("users").where("id", user_id).first()
        return {"status": 200, "body": user}
    except Exception as e:
        return {
            "status": 500,
            "body": {"error": str(e)}
        }

Use rate limiting for public endpoints

@Endpoint(
    id="public_api",
    method="GET",
    path="/public",
    auth=False,
    rate_limit=100  # 100 req/min
)
def public_api() -> dict:
    ...

Next Steps

Database

Store data persistently

Deploying

Deploy your endpoints

Integrations

Use integrations in endpoints

Actions

Build actions for complex logic

Getting Started

Agentic

Application

Integrations

Deployment

Overview

What are Endpoints?

Building Endpoints

Basic Endpoint

Endpoint Parameters

HTTP Methods

GET Endpoint

POST Endpoint

PUT Endpoint

DELETE Endpoint

Path Parameters

Query Parameters

Response Format

Authentication

Require Auth (Default)

Public Endpoint

Rate Limiting

Using Database

Using Integrations

Complete Example: REST API

Best Practices

Next Steps

Database

Deploying

Integrations

Actions

Getting Started

Agentic

Application

Integrations

Deployment

​Overview

​What are Endpoints?

​Building Endpoints

​Basic Endpoint

​Endpoint Parameters

​HTTP Methods

​GET Endpoint

​POST Endpoint

​PUT Endpoint

​DELETE Endpoint

​Path Parameters

​Query Parameters

​Response Format

​Authentication

​Require Auth (Default)

​Public Endpoint

​Rate Limiting

​Using Database

​Using Integrations

​Complete Example: REST API

​Best Practices

​Next Steps

Database

Deploying

Integrations

Actions

Overview

What are Endpoints?

Building Endpoints

Basic Endpoint

Endpoint Parameters

HTTP Methods

GET Endpoint

POST Endpoint

PUT Endpoint

DELETE Endpoint

Path Parameters

Query Parameters

Response Format

Authentication

Require Auth (Default)

Public Endpoint

Rate Limiting

Using Database

Using Integrations

Complete Example: REST API

Best Practices

Next Steps