stream() — anthropic-sdk-python Function Reference

Architecture documentation for the stream() function in messages.py from the anthropic-sdk-python codebase.

Function python AnthropicClient SyncAPI calls 4 called by 1

Entity Profile

AnthropicClient→ SyncAPI→ stream() — anthropic-sdk-python Function Reference

Dependency Diagram

graph TD
  e626c920_4ef0_f218_ed25_ad7210b4668b["stream()"]
  db0dd20b_744d_feba_f585_1c4e91005d64["AsyncMessages"]
  e626c920_4ef0_f218_ed25_ad7210b4668b -->|defined in| db0dd20b_744d_feba_f585_1c4e91005d64
  11f0e821_29f9_1b06_ca3c_9f9f9b07b9c7["stream()"]
  11f0e821_29f9_1b06_ca3c_9f9f9b07b9c7 -->|calls| e626c920_4ef0_f218_ed25_ad7210b4668b
  11f0e821_29f9_1b06_ca3c_9f9f9b07b9c7["stream()"]
  e626c920_4ef0_f218_ed25_ad7210b4668b -->|calls| 11f0e821_29f9_1b06_ca3c_9f9f9b07b9c7
  eedd35ac_a896_dc63_a3fe_c46cf577bffb["_validate_output_config_conflict()"]
  e626c920_4ef0_f218_ed25_ad7210b4668b -->|calls| eedd35ac_a896_dc63_a3fe_c46cf577bffb
  b3ae1af8_b549_4aef_c8c4_9309fd981a1a["_warn_output_format_deprecated()"]
  e626c920_4ef0_f218_ed25_ad7210b4668b -->|calls| b3ae1af8_b549_4aef_c8c4_9309fd981a1a
  618c478a_823f_e214_e25b_6e82a7278cf7["_merge_output_configs()"]
  e626c920_4ef0_f218_ed25_ad7210b4668b -->|calls| 618c478a_823f_e214_e25b_6e82a7278cf7
  style e626c920_4ef0_f218_ed25_ad7210b4668b fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

src/anthropic/resources/beta/messages/messages.py lines 3253–3368

    def stream(
        self,
        *,
        max_tokens: int,
        messages: Iterable[BetaMessageParam],
        model: ModelParam,
        metadata: BetaMetadataParam | Omit = omit,
        output_config: BetaOutputConfigParam | Omit = omit,
        output_format: None | type[ResponseFormatT] | BetaJSONOutputFormatParam | Omit = omit,
        container: Optional[message_create_params.Container] | Omit = omit,
        context_management: Optional[BetaContextManagementConfigParam] | Omit = omit,
        inference_geo: Optional[str] | Omit = omit,
        mcp_servers: Iterable[BetaRequestMCPServerURLDefinitionParam] | Omit = omit,
        service_tier: Literal["auto", "standard_only"] | Omit = omit,
        speed: Optional[Literal["standard", "fast"]] | Omit = omit,
        stop_sequences: SequenceNotStr[str] | Omit = omit,
        system: Union[str, Iterable[BetaTextBlockParam]] | Omit = omit,
        temperature: float | Omit = omit,
        thinking: BetaThinkingConfigParam | Omit = omit,
        tool_choice: BetaToolChoiceParam | Omit = omit,
        tools: Iterable[BetaToolUnionParam] | Omit = omit,
        top_k: int | Omit = omit,
        top_p: float | Omit = omit,
        betas: List[AnthropicBetaParam] | Omit = omit,
        # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
        # The extra values given here take precedence over values defined on the client or passed to this method.
        extra_headers: Headers | None = None,
        extra_query: Query | None = None,
        extra_body: Body | None = None,
        timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
    ) -> BetaAsyncMessageStreamManager[ResponseFormatT]:
        _validate_output_config_conflict(output_config, output_format)
        _warn_output_format_deprecated(output_format)

        if model in DEPRECATED_MODELS:
            warnings.warn(
                f"The model '{model}' is deprecated and will reach end-of-life on {DEPRECATED_MODELS[model]}.\nPlease migrate to a newer model. Visit https://docs.anthropic.com/en/docs/resources/model-deprecations for more information.",
                DeprecationWarning,
                stacklevel=3,
            )

        if model in MODELS_TO_WARN_WITH_THINKING_ENABLED and thinking and thinking["type"] == "enabled":
            warnings.warn(
                f"Using Claude with {model} and 'thinking.type=enabled' is deprecated. Use 'thinking.type=adaptive' instead which results in better model performance in our testing: https://platform.claude.com/docs/en/build-with-claude/adaptive-thinking",
                UserWarning,
                stacklevel=3,
            )

        extra_headers = {
            "X-Stainless-Helper-Method": "stream",
            "X-Stainless-Stream-Helper": "beta.messages",
            **strip_not_given({"anthropic-beta": ",".join(str(e) for e in betas) if is_given(betas) else NOT_GIVEN}),
            **(extra_headers or {}),
        }

        transformed_output_format: BetaJSONOutputFormatParam | Omit = omit

        if is_dict(output_format):
            transformed_output_format = cast(BetaJSONOutputFormatParam, output_format)
        elif is_given(output_format) and output_format is not None:
            adapted_type: TypeAdapter[ResponseFormatT] = TypeAdapter(output_format)

            try:
                schema = adapted_type.json_schema()
                transformed_output_format = BetaJSONOutputFormatParam(
                    schema=transform_schema(schema), type="json_schema"
                )
            except pydantic.errors.PydanticSchemaGenerationError as e:
                raise TypeError(
                    (
                        "Could not generate JSON schema for the given `output_format` type. "
                        "Use a type that works with `pydantic.TypeAdapter`"
                    )
                ) from e

        merged_output_config = _merge_output_configs(output_config, transformed_output_format)

        request = self._post(
            "/v1/messages?beta=true",
            body=maybe_transform(
                {