POST
/
functions
/
{function_id}
/
call
/
stream
from typing import Optional
from pydantic import BaseModel, Field
from opperai import Opper
import time

opper = Opper(http_bearer="YOUR_API_KEY")


class QuestionInput(BaseModel):
    question: str = Field(description="The physics question to explain")
    context: Optional[str] = Field(
        None, description="Additional context about the learner"
    )


# Prepare input data
input_data = QuestionInput(
    question="Explain quantum computing in simple terms",
    context="I'm a beginner in physics",
)

# Stream a response using the direct stream API
unique_name = f"physics_tutor_{int(time.time())}"
stream_response = opper.stream(
    name=unique_name,
    instructions="You are a friendly physics tutor. Explain complex physics concepts in simple, easy-to-understand terms. Use analogies and examples that beginners can relate to.",
    input_schema=QuestionInput,
    input=input_data,
    model="openai/gpt-4o-mini",
)

print("Streaming response:")
# The stream method returns a response object with 'result' containing the EventStream
for event in stream_response.result:
    # Each event is a FunctionStreamCallStreamPostResponseBody with 'data' containing the streaming chunk
    if hasattr(event, "data") and hasattr(event.data, "delta") and event.data.delta:
        print(event.data.delta, end="", flush=True)

print("\nStream completed.")
{
  "data": {
    "delta": "Hello! How can I assist you today?",
    "span_id": "123e4567-e89b-12d3-a456-426614174000"
  }
}

Authorizations

​
Authorization
string
header
required

Bearer authentication header of the form Bearer <token>, where <token> is your auth token.

Path Parameters

​
function_id
string
required

The id of the function to call

Body

application/json

Response

200
text/event-stream

Server-Sent Events stream of function execution chunks

Server-Sent Event following the SSE specification