Recipe Serve (Python)

Programmatic server configuration and management using Python

from praisonaiagents import Agent

agent = Agent(name="code-recipe-agent", instructions="Execute code-based recipes.")
agent.start("Run the code-generation recipe to scaffold a new FastAPI project.")

The user hits the recipe server over HTTP; Python serve() config controls auth, CORS, and ports.

Expose recipe workflows over HTTP — configure auth, CORS, and deployment from Python or YAML.

from praisonai.recipe.serve import serve

serve(host="127.0.0.1", port=8765)

How It Works

Quick Start

1. Simple Usage

   Start a local recipe server:

   from praisonai.recipe.serve import serve
   
   serve(host="127.0.0.1", port=8765)

   Or via CLI:

   praisonai recipe serve --host 127.0.0.1 --port 8765

2. With Configuration

   Load settings from serve.yaml and override auth:

   import os
   from praisonai.recipe.serve import serve, load_config
   
   config = load_config("serve.yaml")
   config["auth"] = "api-key"
   config["api_key"] = os.environ["PRAISONAI_API_KEY"]
   
   serve(
       host=config.get("host", "127.0.0.1"),
       port=config.get("port", 8765),
       config=config,
   )

How It Works

create_app() builds a Starlette ASGI application; serve() runs it with uvicorn. Configuration merges file, environment variables, and Python overrides — CLI flags take highest precedence.

Configuration Options

serve.yaml example

host: 127.0.0.1
port: 8765
auth: api-key
recipes:
  - support-reply-drafter
  - meeting-action-items
preload: true
cors_origins: "https://app.example.com"

Set PRAISONAI_API_KEY in the environment instead of hardcoding keys.

Concurrency model

The recipe server keeps its event loop responsive while long recipes run.

Every recipe.* call (list_recipes, describe, get_schema, run, validate, admin_reload) is dispatched through asyncio.to_thread. A long-running recipe no longer blocks the single event loop — /health, /metrics, and every other request keep responding during execution.

SSE streaming (POST /v1/recipes/stream) drives the underlying sync generator on a worker thread and hands each event back to the loop through an asyncio.Queue (max 64 items). Events reach the client without any next() blocking the loop, and the worker is signalled and drained cleanly on client disconnect so no threads leak.

For operators: a single uvicorn worker can serve mixed short and long recipes without starving readiness probes. You no longer need a separate process per long recipe to keep the health check responsive.

Streaming

praisonai serve recipe streams progressively when fronted by a gateway channel — the same DraftStreamer edits and SSE chunks a Python-Agent-backed bot produces.

The adapter maps each recipe event to a stream event the channel renders:

Subscribe to the adapter’s stream_emitter and drive it with the async surface:

import asyncio
from praisonaiagents.gateway.adapters.recipe_adapter import RecipeBotAdapter

adapter = RecipeBotAdapter("my_recipe")

async def on_event(event):
    print(event.type, getattr(event, "content", ""))

adapter.stream_emitter.add_callback(on_event)

async def main():
    final = await adapter.astart("hello", stream=True)
    print(final)

asyncio.run(main())

Common Patterns

ASGI app for custom deployment

from praisonai.recipe.serve import create_app
import uvicorn

app = create_app(config={"auth": "none", "cors_origins": "*"})
uvicorn.run(app, host="0.0.0.0", port=8765)

Unified agents.yaml config

serve:
  host: 127.0.0.1
  port: 8765
  auth: api-key
  preload: true

from praisonai.recipe.serve import load_config, serve

config = load_config("agents.yaml").get("serve", {})
serve(host=config.get("host", "127.0.0.1"), port=config.get("port", 8765), config=config)

Test with TestClient

import os
from starlette.testclient import TestClient
from praisonai.recipe.serve import create_app

app = create_app(config={"auth": "api-key", "api_key": "test-key"})
client = TestClient(app)

assert client.get("/health").json()["status"] == "healthy"
assert client.get("/v1/recipes", headers={"X-API-Key": "test-key"}).status_code == 200

Best Practices

Related