test_stream_method_definition_in_sync() — anthropic-sdk-python Function Reference

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

Function python AnthropicClient SyncAPI

Entity Profile

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

Dependency Diagram

graph TD
  bb62a388_fb62_5a1d_2fff_ab013913c96b["test_stream_method_definition_in_sync()"]
  8cc8cd20_a242_314f_b59d_2d554c624e7c["test_messages.py"]
  bb62a388_fb62_5a1d_2fff_ab013913c96b -->|defined in| 8cc8cd20_a242_314f_b59d_2d554c624e7c
  style bb62a388_fb62_5a1d_2fff_ab013913c96b fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

tests/lib/streaming/test_messages.py lines 275–303

def test_stream_method_definition_in_sync(sync: bool) -> None:
    client: Anthropic | AsyncAnthropic = sync_client if sync else async_client

    sig = inspect.signature(client.messages.stream)
    generated_sig = inspect.signature(client.messages.create)

    errors: list[str] = []

    for name, generated_param in generated_sig.parameters.items():
        if name == "stream":
            # intentionally excluded
            continue

        custom_param = sig.parameters.get(name)
        if not custom_param:
            errors.append(f"the `{name}` param is missing")
            continue

        if custom_param.annotation != generated_param.annotation:
            errors.append(
                f"types for the `{name}` param are do not match; generated={repr(generated_param.annotation)} custom={repr(custom_param.annotation)}"
            )
            continue

    if errors:
        raise AssertionError(
            f"{len(errors)} errors encountered with the {'sync' if sync else 'async'} client `messages.stream()` method:\n\n"
            + "\n\n".join(errors)
        )