- Rebased SDK MCP implementation onto PR #139's control protocol refactoring
- Moved SDK MCP handling from Transport to Query class for proper layering
- Transport now only filters SDK servers from CLI config
- Query class handles SDK MCP control requests via _handle_sdk_mcp_request
- Added tool decorator and create_sdk_mcp_server API functions
- Added McpSdkServerConfig type definition
- Updated documentation with SDK MCP examples
- Added integration tests for SDK MCP functionality
This implementation properly layers SDK MCP on top of the bidirectional control protocol from PR #139, ensuring clean separation between transport (I/O) and query (protocol) layers.
This PR simplifies the example with interrupt.
There is a flag interrupt_sent that isn't required since the execution
loop will terminate after interrupt with the following message:
`{"content": [{"type": "text", "text": "[Request interrupted by
user]"}]}`
## Key changes
- Adds env field to `ClaudeCodeOptions`, allowing custom env vars to cli
- Updates tests and examples
## Motivation
Bringing Python SDK to feature parity with TS SDK, which supports custom
env vars
## Notes
- Environment variables are merged in order: system env → user env → SDK
required vars
- This implementation seems slightly more robust than the TypeScript
version, which can exclude OS envs vars if a user passes a minimal env
object
- Some linting changes seem to have been picked up
## Summary
- Replace asyncio.create_task() with anyio task group for trio
compatibility
- Update client.py docstring example to use anyio.sleep
- Add trio example demonstrating multi-turn conversation
## Details
The SDK already uses anyio for most async operations, but one line was
using asyncio.create_task() which broke trio compatibility. This PR
fixes that by using anyio's task group API with proper lifecycle
management.
### Changes:
1. **subprocess_cli.py**: Replace asyncio.create_task() with anyio task
group, ensuring proper cleanup on disconnect
2. **client.py**: Update docstring example to use anyio.sleep instead of
asyncio.sleep
3. **streaming_mode_trio.py**: Add new example showing how to use the
SDK with trio
## Test plan
- [x] All existing tests pass
- [x] Manually tested with trio runtime (created test script that
successfully runs multi-turn conversation)
- [x] Linting and type checking pass
🤖 Generated with [Claude Code](https://claude.ai/code)
