stream() — anthropic-sdk-python Function Reference

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

Function python AnthropicClient Authentication calls 1 called by 1

Entity Profile

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

Dependency Diagram

graph TD
  43284ee0_8672_7150_889b_e464fb5a8cf1["stream()"]
  bafd0838_b212_5f78_7082_11123724d722["AsyncMessages"]
  43284ee0_8672_7150_889b_e464fb5a8cf1 -->|defined in| bafd0838_b212_5f78_7082_11123724d722
  f79b127d_5a1b_a7b8_4eda_ad6f6166bae9["stream()"]
  f79b127d_5a1b_a7b8_4eda_ad6f6166bae9 -->|calls| 43284ee0_8672_7150_889b_e464fb5a8cf1
  f79b127d_5a1b_a7b8_4eda_ad6f6166bae9["stream()"]
  43284ee0_8672_7150_889b_e464fb5a8cf1 -->|calls| f79b127d_5a1b_a7b8_4eda_ad6f6166bae9
  style 43284ee0_8672_7150_889b_e464fb5a8cf1 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

src/anthropic/resources/messages/messages.py lines 2423–2533

    def stream(
        self,
        *,
        max_tokens: int,
        messages: Iterable[MessageParam],
        model: ModelParam,
        inference_geo: Optional[str] | Omit = omit,
        metadata: MetadataParam | Omit = omit,
        output_config: OutputConfigParam | Omit = omit,
        output_format: None | JSONOutputFormatParam | type[ResponseFormatT] | Omit = omit,
        container: Optional[str] | Omit = omit,
        service_tier: Literal["auto", "standard_only"] | Omit = omit,
        stop_sequences: SequenceNotStr[str] | Omit = omit,
        system: Union[str, Iterable[TextBlockParam]] | Omit = omit,
        temperature: float | Omit = omit,
        top_k: int | Omit = omit,
        top_p: float | Omit = omit,
        thinking: ThinkingConfigParam | Omit = omit,
        tool_choice: ToolChoiceParam | Omit = omit,
        tools: Iterable[ToolUnionParam] | 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,
    ) -> AsyncMessageStreamManager[ResponseFormatT]:
        """Create a Message stream"""
        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": "messages",
            **(extra_headers or {}),
        }

        transformed_output_format: Optional[JSONOutputFormatParam] | NotGiven = NOT_GIVEN

        if is_dict(output_format):
            transformed_output_format = cast(JSONOutputFormatParam, 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 = JSONOutputFormatParam(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

        # Merge output_format into output_config
        merged_output_config: OutputConfigParam | Omit = omit
        if is_given(transformed_output_format):
            if is_given(output_config):
                merged_output_config = {**output_config, "format": transformed_output_format}
            else:
                merged_output_config = {"format": transformed_output_format}
        elif is_given(output_config):
            merged_output_config = output_config

        request = self._post(
            "/v1/messages",
            body=maybe_transform(
                {
                    "max_tokens": max_tokens,