Async vs Sync (Python only)

Node.js only: The Node.js SDK is async-first. All methods return promises and use await or .then(). Skip this guide if you're using Node.

The Python SDK provides two clients: Noukai (sync) and AsyncNoukai (async). Both have the same API — choose based on your environment.

Sync client (`Noukai`)

Use the sync client for scripts, command-line tools, and Jupyter notebooks:

from noukai_sdk import Noukai
 
# Best for: scripts, CLI tools, Jupyter
with Noukai(org="acme", project="spelling", api_key="nk_...") as client:
    result = client.flow("grade-3").execute(message="Hello")
    print(result.output)

Advantages:

Simple and familiar (no async/await)
Works in Jupyter notebooks
Good for one-off scripts and testing
Automatic resource cleanup via context manager

Limitations:

Cannot iterate (no steps() or events()) — use execute() instead
Blocks the calling thread
Not suitable for high-concurrency servers

Async client (`AsyncNoukai`)

Use the async client for FastAPI applications, aiohttp servers, and long-running async services:

from noukai_sdk import AsyncNoukai
 
# Best for: FastAPI, aiohttp, async services
async def main():
    async with AsyncNoukai(org="acme", project="spelling", api_key="nk_...") as client:
        result = await client.flow("grade-3").execute(message="Hello")
        print(result.output)
 
import asyncio
asyncio.run(main())

Advantages:

Full iteration support via steps() and events()
Non-blocking (can handle thousands of concurrent flows)
Perfect for async frameworks like FastAPI, aiohttp, asyncio
Automatic resource cleanup via async context manager

Limitations:

Requires async/await syntax
Must be called from an async context

FastAPI integration

Use AsyncNoukai in FastAPI endpoints for non-blocking flow execution:

from fastapi import FastAPI
from noukai_sdk import AsyncNoukai
 
app = FastAPI()
 
@app.post("/grade")
async def grade_essay(essay: str):
    async with AsyncNoukai(api_key="nk_...") as client:
        result = await client.flow("acme/spelling/grade-3").execute(
            message=essay
        )
        return {"output": result.output}

For streaming responses, iterate with events():

from fastapi.responses import StreamingResponse
 
@app.post("/grade-stream")
async def grade_essay_stream(essay: str):
    async def event_generator():
        async with AsyncNoukai(org="acme", project="spelling", api_key="nk_...") as client:
            flow = client.flow("grade-3")
            async for event in flow.events(message=essay):
                if hasattr(event, "output"):
                    yield f"data: {event.output}\n\n"
    
    return StreamingResponse(event_generator(), media_type="text/event-stream")

Jupyter notebooks

Use sync client (Noukai) in Jupyter for simplicity:

from noukai_sdk import Noukai
 
# In a Jupyter cell
with Noukai(org="acme", project="spelling", api_key="nk_...") as client:
    result = client.flow("grade-3").execute(message="Hello")
    print(result.output)

For async in Jupyter, use the ipython event loop:

from noukai_sdk import AsyncNoukai
import nest_asyncio
 
# Enable nested event loops in Jupyter
nest_asyncio.apply()
 
async def main():
    async with AsyncNoukai(org="acme", project="spelling", api_key="nk_...") as client:
        result = await client.flow("grade-3").execute(message="Hello")
        return result.output
 
# Run async function in Jupyter
result = await main()
print(result)

Converting between sync and async

If you have sync code but need async iteration, wrap it with asyncio.run():

import asyncio
 
async def async_iterate():
    async with AsyncNoukai(org="acme", project="spelling", api_key="nk_...") as client:
        flow = client.flow("grade-3")
        async for step in flow.steps(message="hello"):
            print(step.output)
 
# Run in sync context
asyncio.run(async_iterate())

For the reverse (async code in sync), use libraries like nest_asyncio (Jupyter) or asgiref (rare):

# Not recommended — use AsyncNoukai directly in async contexts
from asgiref.sync import async_to_sync
 
async def fetch_flow():
    async with AsyncNoukai(org="acme", project="spelling", api_key="nk_...") as client:
        return await client.flow("grade-3").execute(message="hi")
 
# Call from sync code (not ideal — blocks the thread)
result = async_to_sync(fetch_flow)()

Resource management

Both clients are context managers and should be used with with or async with to ensure proper cleanup:

# Sync
with Noukai() as client:
    # Use client
    pass
# Resources cleaned up automatically
 
# Async
async with AsyncNoukai() as client:
    # Use client
    pass
# Resources cleaned up automatically

Or manage manually:

# Sync
client = Noukai()
try:
    result = client.flow("...").execute(message="...")
finally:
    client.close()
 
# Async
client = AsyncNoukai()
try:
    result = await client.flow("...").execute(message="...")
finally:
    await client.aclose()

Concurrency limits

For high-concurrency scenarios (e.g., FastAPI handling 1000s of concurrent requests), use AsyncNoukai:

# FastAPI with 100 concurrent flows
from fastapi import FastAPI
from noukai_sdk import AsyncNoukai
 
app = FastAPI()
 
@app.post("/process")
async def process(data: dict):
    async with AsyncNoukai(org="acme", project="process") as client:
        # This runs concurrently with other requests
        result = await client.flow("default").execute(**data)
        return result

Using sync client in a thread pool is an alternative but less efficient:

from concurrent.futures import ThreadPoolExecutor
 
with ThreadPoolExecutor(max_workers=10) as executor:
    # Blocks threads, not recommended
    result = executor.submit(sync_flow).result()

Choose the right client

Use case	Client	Why
Scripts, CLI, Jupyter	`Noukai`	Simple, no async overhead
FastAPI endpoints	`AsyncNoukai`	Non-blocking, scales
Streaming responses	`AsyncNoukai`	Required for `events()` iteration
High concurrency	`AsyncNoukai`	Handles 1000s of concurrent flows
Legacy code	`Noukai`	No refactoring needed

Next steps

Tool calls: See Tool calls
Error handling: See Errors & retries

Async vs Sync (Python only)

On this page