Skip to content

API Reference

Client

openresponses.client.OpenResponsesClient

Async/Sync Client for Open Responses API.

Source code in src/openresponses/client.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class OpenResponsesClient:
    """
    Async/Sync Client for Open Responses API.
    """
    def __init__(self, base_url: str, api_key: Optional[str] = None):
        self.base_url = base_url.rstrip("/")
        self.headers = {"Content-Type": "application/json"}
        if api_key:
            self.headers["Authorization"] = f"Bearer {api_key}"

    def create(
        self,
        model: str,
        input: Union[str, List[MessageItem]],
        stream: bool = False,
        max_tool_calls: Optional[int] = None
    ) -> Union[OpenResponsesOutput, Generator[StreamEvent, None, None]]:
        """
        Synchronous request to create a response.
        """
        request = OpenResponsesRequest(
            model=model,
            input=input,
            stream=stream,
            max_tool_calls=max_tool_calls
        )

        url = f"{self.base_url}/v1/responses"

        if stream:
            return self._stream_request(url, request)
        else:
            with httpx.Client() as client:
                resp = client.post(url, json=request.model_dump(), headers=self.headers, timeout=60.0)
                resp.raise_for_status()
                return OpenResponsesOutput(**resp.json())

    def _stream_request(self, url: str, request: OpenResponsesRequest) -> Generator[StreamEvent, None, None]:
        with httpx.Client() as client:
            with client.stream("POST", url, json=request.model_dump(), headers=self.headers, timeout=60.0) as resp:
                resp.raise_for_status()
                for line in resp.iter_lines():
                    if line.startswith("event:"):
                        event_type = line.split(": ", 1)[1]
                    elif line.startswith("data:"):
                        data_str = line.split(": ", 1)[1]
                        data = json.loads(data_str)
                        yield StreamEvent(event=event_type, data=data)

create(model, input, stream=False, max_tool_calls=None)

Synchronous request to create a response.

Source code in src/openresponses/client.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def create(
    self,
    model: str,
    input: Union[str, List[MessageItem]],
    stream: bool = False,
    max_tool_calls: Optional[int] = None
) -> Union[OpenResponsesOutput, Generator[StreamEvent, None, None]]:
    """
    Synchronous request to create a response.
    """
    request = OpenResponsesRequest(
        model=model,
        input=input,
        stream=stream,
        max_tool_calls=max_tool_calls
    )

    url = f"{self.base_url}/v1/responses"

    if stream:
        return self._stream_request(url, request)
    else:
        with httpx.Client() as client:
            resp = client.post(url, json=request.model_dump(), headers=self.headers, timeout=60.0)
            resp.raise_for_status()
            return OpenResponsesOutput(**resp.json())

options: show_root_heading: true

openresponses.client.AsyncOpenResponsesClient

Async Client for Open Responses API.

Source code in src/openresponses/client.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
class AsyncOpenResponsesClient:
    """
    Async Client for Open Responses API.
    """
    def __init__(self, base_url: str, api_key: Optional[str] = None):
        self.base_url = base_url.rstrip("/")
        self.headers = {"Content-Type": "application/json"}
        if api_key:
            self.headers["Authorization"] = f"Bearer {api_key}"

    async def create(
        self,
        model: str,
        input: Union[str, List[MessageItem]],
        stream: bool = False,
        max_tool_calls: Optional[int] = None
    ) -> Union[OpenResponsesOutput, AsyncGenerator[StreamEvent, None]]:
        request = OpenResponsesRequest(
            model=model,
            input=input,
            stream=stream,
            max_tool_calls=max_tool_calls
        )

        url = f"{self.base_url}/v1/responses"

        if stream:
            return self._stream_request(url, request)
        else:
            async with httpx.AsyncClient() as client:
                resp = await client.post(url, json=request.model_dump(), headers=self.headers, timeout=60.0)
                resp.raise_for_status()
                return OpenResponsesOutput(**resp.json())

    async def _stream_request(self, url: str, request: OpenResponsesRequest) -> AsyncGenerator[StreamEvent, None]:
        async with httpx.AsyncClient() as client:
            async with client.stream("POST", url, json=request.model_dump(), headers=self.headers, timeout=60.0) as resp:
                resp.raise_for_status()
                async for line in resp.aiter_lines():
                    if line.startswith("event:"):
                        event_type = line.split(": ", 1)[1]
                    elif line.startswith("data:"):
                        data_str = line.split(": ", 1)[1]
                        # Handle potential keeping of newlines or empty data
                        try:
                            data = json.loads(data_str)
                            yield StreamEvent(event=event_type, data=data)
                        except json.JSONDecodeError:
                            continue

options: show_root_heading: true

Models

openresponses.models.OpenResponsesRequest

Bases: BaseModel

Standard Request Body for Open Responses API.

Source code in src/openresponses/models.py 
47
48
49
50
51
52
class OpenResponsesRequest(BaseModel):
    """Standard Request Body for Open Responses API."""
    model: str
    input: Union[str, List[MessageItem]] # Can be simple text or structured items
    stream: bool = False
    max_tool_calls: Optional[int] = Field(default=None, description="Limit for provider-managed loops")

options: members: - model - input - stream - max_tool_calls

openresponses.models.OpenResponsesOutput

Bases: BaseModel

Standard Response Body for Open Responses API.

Source code in src/openresponses/models.py 
56
57
58
59
60
61
62
class OpenResponsesOutput(BaseModel):
    """Standard Response Body for Open Responses API."""
    id: str
    object: Literal["response"] = "response"
    created: int
    model: str
    output: List[ResponseItem]

openresponses.models.MessageItem

Bases: BaseModel

Represents a standard chat message.

Source code in src/openresponses/models.py 
20
21
22
23
24
class MessageItem(BaseModel):
    """Represents a standard chat message."""
    type: Literal["message"] = "message"
    role: Literal["user", "assistant", "system"]
    content: Union[str, List[InputText]]

openresponses.models.ReasoningItem

Bases: BaseModel

Represents a block of reasoning/thought process.

Source code in src/openresponses/models.py 
28
29
30
31
32
33
class ReasoningItem(BaseModel):
    """Represents a block of reasoning/thought process."""
    type: Literal["reasoning"] = "reasoning"
    content: Optional[str] = Field(default=None, description="Raw reasoning traces (Transparent)")
    summary: Optional[str] = Field(default=None, description="Sanitized summary")
    encrypted_content: Optional[str] = Field(default=None, description="Provider-secure reasoning")

openresponses.models.ToolCallItem

Bases: BaseModel

Represents a tool call request.

Source code in src/openresponses/models.py 
35
36
37
38
39
40
class ToolCallItem(BaseModel):
    """Represents a tool call request."""
    type: Literal["tool_call"] = "tool_call"
    id: str
    name: str
    arguments: Dict[str, Any]