Skip to content

Comments

feat: Add sync streaming support for Anthropic instrumentation#4155

Open
vasantteja wants to merge 36 commits intoopen-telemetry:mainfrom
vasantteja:anthropic-sync-streaming
Open

feat: Add sync streaming support for Anthropic instrumentation#4155
vasantteja wants to merge 36 commits intoopen-telemetry:mainfrom
vasantteja:anthropic-sync-streaming

Conversation

@vasantteja
Copy link
Contributor

@vasantteja vasantteja commented Feb 1, 2026

Description

This PR adds sync streaming support for the Anthropic instrumentation. It enables telemetry capture for:

  1. Messages.create(stream=True) - Streaming responses via the create method with stream parameter
  2. Messages.stream() - The dedicated streaming method that returns a MessageStreamManager

Key changes:

  • Added StreamWrapper class to wrap Stream[RawMessageStreamEvent] and extract telemetry from streaming chunks
  • Added MessageStreamManagerWrapper to wrap MessageStreamManager context manager
  • Added MessageWrapper for non-streaming response telemetry extraction
  • Renamed MessageCreateParams to MessageRequestParams to reflect broader API coverage
  • Modified messages_create to use manual lifecycle management (start_llm/stop_llm) instead of context manager to support both streaming and non-streaming

Fixes #3949 partially.

Type of change

Please delete options that are not relevant.

  • Bug fix (non-breaking change which fixes an issue)
  • New feature (non-breaking change which adds functionality)
  • Breaking change (fix or feature that would cause existing functionality to not work as expected)
  • This change requires a documentation update

How Has This Been Tested?

Added comprehensive tests for sync streaming functionality:

  • test_sync_messages_create_streaming - Tests streaming with context manager
  • test_sync_messages_create_streaming_iteration - Tests direct iteration without context manager
  • test_sync_messages_create_streaming_connection_error - Tests error handling for streaming
  • test_sync_messages_stream_basic - Tests Messages.stream() method
  • test_sync_messages_stream_with_params - Tests stream with additional parameters (temperature, top_p, top_k)
  • test_sync_messages_stream_token_usage - Tests token usage capture in streaming
  • test_sync_messages_stream_connection_error - Tests error handling for stream method

All tests use VCR cassettes for reproducible HTTP interaction replay.

Does This PR Require a Core Repo Change?

  • Yes. - Link to PR:
  • No.

Checklist:

See contributing.md for styleguide, changelog guidelines, and more.

  • Followed the style guidelines of this project
  • Changelogs have been updated
  • Unit tests have been added
  • Documentation has been updated

- Add support for Messages.create(stream=True) with StreamWrapper
- Add support for Messages.stream() with MessageStreamManagerWrapper
- Add MessageWrapper for non-streaming response telemetry
- Rename MessageCreateParams to MessageRequestParams
- Add comprehensive tests for sync streaming functionality
- Add type: ignore[arg-type] for Union type narrowing in messages_create
- Add type: ignore[return-value] for wrapper return types
- Add type: ignore[return-value] for __exit__ returning None
@vasantteja vasantteja force-pushed the anthropic-sync-streaming branch from 99a2596 to 504d0df Compare February 1, 2026 16:57
@vasantteja vasantteja removed their assignment Feb 5, 2026
@lmolkova
Copy link
Member

lmolkova commented Feb 8, 2026

tagging @anirudha who was interested to review the PR :)

@anirudha
Copy link

anirudha commented Feb 8, 2026

Thanks. Taking a look today

…r handling

- Introduce constants for provider name and cache token attributes.
- Normalize stop reasons and aggregate cache token fields in MessageWrapper and StreamWrapper.
- Enhance tests to validate input token aggregation and stop reason normalization.
- Update cassettes for new request and response structures in streaming scenarios.
@vasantteja vasantteja removed their assignment Feb 9, 2026
…d consistency

- Simplify constant definitions and normalize function calls in utils.py.
- Enhance test cases by removing unnecessary line breaks and improving formatting.
- Ensure consistent usage of type hints and comments in test functions.
@vasantteja vasantteja removed their assignment Feb 9, 2026
- Update the pylint directive to disable too-many-arguments warning for better clarity.
- Maintain consistency in function signature and improve code readability.
Copy link

@anirudha anirudha left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Tests all pass locally. Nice work overall — the wrapper separation is clean. One bug to fix (double finalize), rest are suggestions.

Note: conftest.py isn't in this diff so I can't leave a line comment, but scrub_response_headers is a no-op and all new cassettes leak anthropic-organization-id: 455ea6be-bd92-4199-83ec-0c6b39c5c169. Worth scrubbing that or adding it to filter_headers.

Also, the PR description says Fixes #3949 but async streaming isn't covered. Totally fine to scope this to sync only, but Fixes will auto-close the issue on merge. Maybe Partially addresses #3949 instead?

…tion

- Update test cases to validate streaming behavior with various parameters, including token usage and stop reasons.
- Introduce new cassettes for different scenarios, ensuring comprehensive coverage of streaming interactions.
- Refactor existing tests for clarity and consistency in structure and assertions.
…ocals in test_stream_wrapper_finalize_idempotent function
@vasantteja vasantteja removed their assignment Feb 16, 2026
…e. Introduced MessageWrapper and StreamWrapper classes for telemetry handling. Updated tests to reflect changes in instrumentation behavior.
…ty functions, and update wrapper classes for better clarity and maintainability. Removed unused code and improved type safety in utility functions. Updated tests to reflect changes in the instrumentation behavior.
…imports and streamline finish reason normalization for improved clarity and maintainability.
@xrmx xrmx requested review from Cirilla-zmh and anirudha February 17, 2026 08:21
Comment on lines +134 to +143
def _skip_if_cassette_missing_and_no_real_key(request):
cassette_path = (
Path(__file__).parent / "cassettes" / f"{request.node.name}.yaml"
)
api_key = os.getenv("ANTHROPIC_API_KEY")
if not cassette_path.exists() and api_key == "test_anthropic_api_key":
pytest.skip(
f"Cassette {cassette_path.name} is missing. "
"Set a real ANTHROPIC_API_KEY to record it."
)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why would the cassette be missing?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The _skip_if_cassette_missing_and_no_real_key guard is called only by the tool_use and thinking tests — whose cassettes may not exist if the SDK version is too old to record them. It prevents a confusing failure (dummy key hitting the real API) by cleanly skipping the test with a message to set a real key. Lmk if you want to remove this.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why are there several new requests in here? It doesn't look like test_sync_messages_create_stop_reason was changed to make more request

In case you haven't seen it, you can pass VCR record mode with --vcr-record=<mode>

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Honeslty thanks for this. I was using this pytest instrumentation-genai/opentelemetry-instrumentation-anthropic/tests --vcr-record=all and it was appending my requests. I was rerunning this everytime I make a change just to ensure everything is good. This bulked up the cassettes. I deleted the cassettes and recreated them.

@vasantteja vasantteja removed their assignment Feb 17, 2026
…st and response structures, enhance error handling scenarios, and ensure consistency in message formats across various test cases. Removed outdated data and improved clarity in test interactions.
@vasantteja vasantteja removed their assignment Feb 18, 2026
Copy link
Member

@MikeGoldsmith MikeGoldsmith left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looking good, thanks @vasantteja.

I've left some suggestions and things we need to figure out if we want to remove or update.

…apper to include content capture logic, improve type safety with explicit casting, and streamline test cases for better clarity. Added new test for streaming response attributes and refined existing tests to ensure consistency in message handling.
…Ds, timestamps, and token usage across various test cases. Refine content capture logic and ensure consistency in message formats, including adjustments to event data and headers for improved clarity and accuracy.
Copy link
Member

@MikeGoldsmith MikeGoldsmith left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🚀

def __enter__(self) -> "StreamWrapper":
return self

def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> bool:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Potential bug: __next__ can call fail_llm(...) on stream errors, but _finalized is not set on that path, and __exit__ always calls close() which then calls _finalize_invocation() -> stop_llm(...). That can finalize the same invocation twice (fail then stop) and potentially produce incorrect success telemetry after an error. Also, because __exit__ ignores exc_type, exceptions raised by user code inside with stream: are currently treated as successful completion.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed!! This abstraction is little error prone. I rewrote it to mimic openai responses stream wrapper incorporating your other suggestion on that PR.


if result.model:
invocation.response_model_name = result.model
is_streaming = kwargs.get("stream", False)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it possible to make this type safe by moving it into the extract_params() function's return value?

…oved type safety. Replace 'Any' with 'object' in several function signatures and class attributes. Introduce logging for error handling in MessagesStreamWrapper to enhance instrumentation reliability.
… clarity and safety. Update function signatures to use specific types instead of 'object', including changes to parameters in extract_params, get_input_messages, and get_system_instruction. Refactor messages_create to ensure correct type handling for streaming and non-streaming responses. Additionally, streamline message handling in MessagesStreamWrapper for better performance and reliability.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Add OpenTelemetry instrumentation for the Anthropic Claude Python SDK