|
34 | 34 | ) |
35 | 35 | from ..models.bidirectional_model import BidiModel |
36 | 36 | from ..models.novasonic import BidiNovaSonicModel |
| 37 | +from ..types.agent import BidiAgentInput |
37 | 38 | from ..types.events import BidiAudioInputEvent, BidiImageInputEvent, BidiTextInputEvent, BidiInputEvent, BidiOutputEvent |
38 | 39 | from ..types import BidiIO |
39 | 40 | from ....experimental.tools import ToolProvider |
|
42 | 43 |
|
43 | 44 | _DEFAULT_AGENT_NAME = "Strands Agents" |
44 | 45 | _DEFAULT_AGENT_ID = "default" |
45 | | -# Type alias for cleaner send() method signature |
46 | | -BidirectionalInput = str | BidiAudioInputEvent | BidiImageInputEvent |
47 | 46 |
|
48 | 47 |
|
49 | 48 | class BidiAgent: |
@@ -256,7 +255,7 @@ async def start(self) -> None: |
256 | 255 | logger.debug("Conversation start - initializing connection") |
257 | 256 | self._agent_loop = await start_bidirectional_connection(self) |
258 | 257 |
|
259 | | - async def send(self, input_data: BidirectionalInput) -> None: |
| 258 | + async def send(self, input_data: BidiAgentInput) -> None: |
260 | 259 | """Send input to the model (text, audio, image, or event dict). |
261 | 260 | |
262 | 261 | Unified method for sending text, audio, and image input to the model during |
@@ -303,24 +302,24 @@ async def send(self, input_data: BidirectionalInput) -> None: |
303 | 302 | if isinstance(input_data, dict) and "type" in input_data: |
304 | 303 | event_type = input_data["type"] |
305 | 304 | if event_type == "bidi_text_input": |
306 | | - input_data = BidiTextInputEvent(text=input_data["text"], role=input_data["role"]) |
| 305 | + input_event = BidiTextInputEvent(text=input_data["text"], role=input_data["role"]) |
307 | 306 | elif event_type == "bidi_audio_input": |
308 | | - input_data = BidiAudioInputEvent( |
| 307 | + input_event = BidiAudioInputEvent( |
309 | 308 | audio=input_data["audio"], |
310 | 309 | format=input_data["format"], |
311 | 310 | sample_rate=input_data["sample_rate"], |
312 | 311 | channels=input_data["channels"] |
313 | 312 | ) |
314 | 313 | elif event_type == "bidi_image_input": |
315 | | - input_data = BidiImageInputEvent( |
| 314 | + input_event = BidiImageInputEvent( |
316 | 315 | image=input_data["image"], |
317 | 316 | mime_type=input_data["mime_type"] |
318 | 317 | ) |
319 | 318 | else: |
320 | 319 | raise ValueError(f"Unknown event type: {event_type}") |
321 | 320 |
|
322 | 321 | # Send the reconstructed TypedEvent |
323 | | - await self._agent_loop.model.send(input_data) |
| 322 | + await self._agent_loop.model.send(input_event) |
324 | 323 | return |
325 | 324 |
|
326 | 325 | # If we get here, input type is invalid |
|
0 commit comments