Implementation:Anthropics Anthropic sdk python JSONL Decoder

Knowledge Sources	Anthropic SDK Python
Domains	Streaming, Data_Parsing
Last Updated	2026-02-15 12:00 GMT

Overview

The JSONLDecoder and AsyncJSONLDecoder classes provide synchronous and asynchronous iterators that parse JSON Lines (newline-delimited JSON) byte streams into typed Python objects.

Description

The JSONL (JSON Lines) format is a text format where each line contains a valid JSON value, separated by newline characters. These decoder classes wrap a raw byte iterator (from an HTTP response body) and yield parsed, type-constructed Python objects for each line.

Both decoders follow the same algorithm: they accumulate bytes into a buffer, split on line boundaries, and when a complete line ending (\r, \n, or \r\n) is detected, they parse the accumulated buffer with json.loads() and construct a typed object using construct_type_unchecked(). After the byte stream is exhausted, any remaining data in the buffer is flushed as a final line. This approach handles chunked transfer encoding where line boundaries may not align with chunk boundaries.

The JSONLDecoder implements the synchronous iterator protocol (__iter__ / __next__), while AsyncJSONLDecoder implements the asynchronous iterator protocol (__aiter__ / __anext__). Both hold a reference to the originating httpx.Response object and provide a close() method to terminate the stream early.

Usage

These decoders are used internally by the SDK when processing API responses that return data in JSONL streaming format. They are not typically instantiated directly by end users.

Code Reference

Source Location

• Repository: Anthropic SDK Python
• File: src/anthropic/_decoders/jsonl.py

Signature

class JSONLDecoder(Generic[_T]):
    http_response: httpx.Response

    def __init__(
        self,
        *,
        raw_iterator: Iterator[bytes],
        line_type: type[_T],
        http_response: httpx.Response,
    ) -> None: ...

    def close(self) -> None: ...
    def __decode__(self) -> Iterator[_T]: ...
    def __next__(self) -> _T: ...
    def __iter__(self) -> Iterator[_T]: ...


class AsyncJSONLDecoder(Generic[_T]):
    http_response: httpx.Response

    def __init__(
        self,
        *,
        raw_iterator: AsyncIterator[bytes],
        line_type: type[_T],
        http_response: httpx.Response,
    ) -> None: ...

    async def close(self) -> None: ...
    async def __decode__(self) -> AsyncIterator[_T]: ...
    async def __anext__(self) -> _T: ...
    async def __aiter__(self) -> AsyncIterator[_T]: ...

Import

# Internal SDK usage
from anthropic._decoders.jsonl import JSONLDecoder, AsyncJSONLDecoder

I/O Contract

Constructor Parameters

Parameter	Type	Description
raw_iterator	Iterator[bytes] / AsyncIterator[bytes]	Raw byte chunks from the HTTP response body stream
line_type	type[_T]	The target type to construct each decoded JSON line into
http_response	httpx.Response	The originating HTTP response, retained for lifecycle management

Output

Output	Type	Description
Yielded items	_T	Each JSON line parsed and constructed into the specified line_type

Decoding Algorithm

def __decode__(self) -> Iterator[_T]:
    buf = b""
    for chunk in self._raw_iterator:
        for line in chunk.splitlines(keepends=True):
            buf += line
            if buf.endswith((b"\r", b"\n", b"\r\n")):
                yield construct_type_unchecked(
                    value=json.loads(buf),
                    type_=self._line_type,
                )
                buf = b""

    # flush remaining data
    if buf:
        yield construct_type_unchecked(
            value=json.loads(buf),
            type_=self._line_type,
        )

Usage Examples

# Internal SDK usage -- synchronous
from anthropic._decoders.jsonl import JSONLDecoder

decoder = JSONLDecoder(
    raw_iterator=response.iter_bytes(),
    line_type=MyResponseType,
    http_response=response,
)

for item in decoder:
    print(item)  # Each item is a MyResponseType instance

# Close early if not consuming the full stream
decoder.close()


# Internal SDK usage -- asynchronous
from anthropic._decoders.jsonl import AsyncJSONLDecoder

async_decoder = AsyncJSONLDecoder(
    raw_iterator=response.aiter_bytes(),
    line_type=MyResponseType,
    http_response=response,
)

async for item in async_decoder:
    print(item)

await async_decoder.close()

Dependencies

• json -- Standard library JSON parsing
• httpx -- HTTP response type for lifecycle management
• anthropic._models.construct_type_unchecked -- Type-aware object construction from parsed JSON data

Related Pages

Related Implementations

• Implementation:Anthropics_Anthropic_sdk_python_Stream_Event_Processing -- Stream event processing that may use JSONL decoding
• Implementation:Anthropics_Anthropic_sdk_python_MessageStream_Iterator -- Message stream iteration