How the SSE Builder in free-claude-code Converts Provider Responses to Anthropic SSE Format

The SSEBuilder class normalizes raw streaming chunks from any provider into Anthropic's Server-Sent Events format by managing message lifecycles, content blocks, and tool calls through a stateful conversion layer.

The Alishahryar1/free-claude-code repository implements an abstraction layer that allows diverse AI providers to communicate through a unified Anthropic-compatible streaming interface. At the heart of this system sits the SSEBuilder class in providers/common/sse_builder.py, which transforms fragmented provider responses into structured Server-Sent Events that adhere to Anthropic's exact SSE contract.

SSEBuilder Architecture and Message Lifecycle

Initializing the Message Stream

The SSEBuilder class orchestrates conversion by first establishing message boundaries. When a stream begins, message_start() generates the initial event containing the message ID, model name, and input token usage stub. At completion, message_delta() and message_stop() emit the corresponding terminal events to signal stream conclusion.

Mapping Provider Stop Reasons

The builder translates provider-specific finish reasons into Anthropic-compatible stop_reason values through the map_stop_reason() method. This mapping converts provider signals like stop to end_turn and length to max_tokens, ensuring downstream clients receive standardized termination indicators regardless of the upstream source.

Content Block Management and Delta Streaming

The ContentBlockManager State Machine

Underlying the SSEBuilder, the ContentBlockManager class tracks open content blocks, allocates monotonic indices, and manages temporary state for tool calls. This component handles fragmented tool names that arrive piece-by-piece from providers and buffers partial JSON arguments until they form valid objects.

Structured Content Block Events

Anthropic streams organize output into indexed content blocks for thinking, text, and tool_use. The SSEBuilder provides specific lifecycle methods for each type:

• content_block_start() opens a block with appropriate payload fields (thinking, text, or tool_use)
• content_block_delta() streams incremental updates via thinking_delta, text_delta, or input_json_delta
• content_block_stop() closes the block with proper indexing

High-level helpers like start_thinking_block(), emit_text_delta(), and start_tool_block() wrap these primitives for ergonomic usage within provider adapters.

Handling Task Tool Arguments

For specialized Task tools, the buffer_task_args() method accumulates partial JSON arguments until complete. Per the Anthropic contract, it automatically sets run_in_background=False before emitting a single consolidated input_json_delta event, preventing premature execution of incomplete tool calls.

Converting OpenAI-Compatible Provider Streams

The _stream_response_impl Pipeline

In providers/openai_compat.py, the asynchronous _stream_response_impl() method consumes raw provider chunks and drives the SSEBuilder accordingly. This implementation handles the three primary content types encountered in OpenAI-compatible streaming responses, with similar patterns appearing in providers/open_router/client.py, providers/llamacpp/client.py, and providers/lmstudio/client.py.

Reasoning and Thinking Content

When providers supply reasoning through reasoning_content fields or THINKING tokens (detected via ThinkTagParser in providers/common/think_parser.py), the code ensures a thinking block is active through ensure_thinking_block() before emitting thinking_delta events. This separates internal reasoning from final output while maintaining the Anthropic block structure.

Text Fragment Processing

Normal text fragments pass through HeuristicToolParser (from heuristic_tool_parser.py) to strip tool syntax artifacts. The ensure_text_block() method opens text blocks when content arrives, with emit_text_delta() streaming sanitized fragments to the client.

Native Tool Call Streaming

Upon detecting delta.tool_calls, the implementation first closes any open text or thinking blocks via close_content_blocks(). For each tool call, it invokes SSEBuilder.start_tool_block() with the tool index and ID, then streams JSON argument deltas through emit_tool_delta(). The system handles fragmented tool names and buffers Task tool arguments until the JSON object is complete.

Error Handling and Recovery

If exceptions occur during streaming, the builder closes open content blocks and emits a synthetic text block containing the error message via emit_error(). This prevents hung streams and ensures clients receive explanatory feedback even when upstream providers fail.

Output Token Estimation

After stream completion, SSEBuilder.estimate_output_tokens() calculates approximate token counts using the optional tiktoken encoder or a fallback heuristic. This method analyzes accumulated text, reasoning content, and buffered tool arguments to provide usage statistics compatible with Anthropic's billing and monitoring expectations.

Practical Implementation Example

The following example demonstrates manual SSEBuilder usage, mirroring the automated conversion performed in providers/openai_compat.py:

from providers.common.sse_builder import SSEBuilder

# Initialise a builder for a new message

sse = SSEBuilder(message_id="msg_123", model="gpt-4o-mini", input_tokens=42)

# Header events

print(sse.message_start())

# Stream a thinking chunk

print(sse.start_thinking_block())
print(sse.emit_thinking_delta("Analyzing the request…"))
print(sse.stop_thinking_block())

# Stream normal text

print(sse.start_text_block())
print(sse.emit_text_delta("Here is the answer you asked for."))
print(sse.stop_text_block())

# Stream a tool call (Task tool – arguments are buffered until complete)

print(sse.start_tool_block(tool_index=0, tool_id="tool_1", name="Task"))
print(sse.emit_tool_delta(0, '{"action":"search","query":"weather"}'))  # full JSON emitted

print(sse.stop_tool_block(0))

# Footer events

print(sse.message_delta(stop_reason="end_turn", output_tokens=15))
print(sse.message_stop())

Summary

• The SSEBuilder class in providers/common/sse_builder.py serves as the central conversion engine for Anthropic-compatible streaming
• ContentBlockManager handles stateful block tracking, index allocation, and Task tool argument buffering
• Provider-specific stop reasons map to Anthropic values like end_turn and max_tokens via map_stop_reason()
• The providers/openai_compat.py adapter demonstrates real-time conversion of thinking content, text deltas, and tool calls
• Token estimation uses tiktoken or fallback heuristics on accumulated content for usage reporting

Frequently Asked Questions

How does the SSEBuilder handle fragmented tool arguments?

The ContentBlockManager.buffer_task_args() method accumulates partial JSON fragments until they form a valid object. For Task tools specifically, it buffers arguments and forces run_in_background=False before emitting a single input_json_delta event, ensuring the Anthropic contract receives complete, executable tool calls rather than streaming JSON fragments.

What is the difference between content_block_delta and the high-level emit methods?

While content_block_delta() is the low-level primitive for generating SSE events, high-level methods like emit_text_delta() and emit_thinking_delta() provide ergonomic wrappers that manage block state automatically. These helpers ensure the appropriate block type is open before emitting deltas, handling the boilerplate of content_block_start() and content_block_stop() calls internally.

How are provider stop reasons converted to Anthropic format?

The map_stop_reason() method translates provider-specific finish reasons into Anthropic-standard values. For example, a provider's stop maps to Anthropic's end_turn, while length maps to max_tokens. This translation occurs before message_delta() emits the final event, ensuring downstream clients receive standardized termination signals regardless of the upstream provider.

Can the SSEBuilder recover from errors during streaming?

Yes, the error handling implementation in providers/openai_compat.py catches exceptions during the _stream_response_impl() loop. When errors occur, the builder first closes any open content blocks via close_content_blocks(), then emits a synthetic text block containing the error message through emit_error(). This guarantees the SSE stream terminates gracefully with explanatory content rather than abruptly disconnecting.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}