Skip to content

strands.multiagent.a2a.server

A2A-compatible wrapper for Strands Agent.

This module provides the A2AAgent class, which adapts a Strands Agent to the A2A protocol, allowing it to be used in A2A-compatible systems.

logger = logging.getLogger(__name__) module-attribute

A2AServer

A2A-compatible wrapper for Strands Agent.

Source code in strands/multiagent/a2a/server.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
class A2AServer:
    """A2A-compatible wrapper for Strands Agent."""

    def __init__(
        self,
        agent: SAAgent,
        *,
        # AgentCard
        host: str = "127.0.0.1",
        port: int = 9000,
        http_url: str | None = None,
        serve_at_root: bool = False,
        version: str = "0.0.1",
        skills: list[AgentSkill] | None = None,
        # RequestHandler
        task_store: TaskStore | None = None,
        queue_manager: QueueManager | None = None,
        push_config_store: PushNotificationConfigStore | None = None,
        push_sender: PushNotificationSender | None = None,
    ):
        """Initialize an A2A-compatible server from a Strands agent.

        Args:
            agent: The Strands Agent to wrap with A2A compatibility.
            host: The hostname or IP address to bind the A2A server to. Defaults to "127.0.0.1".
            port: The port to bind the A2A server to. Defaults to 9000.
            http_url: The public HTTP URL where this agent will be accessible. If provided,
                this overrides the generated URL from host/port and enables automatic
                path-based mounting for load balancer scenarios.
                Example: "http://my-alb.amazonaws.com/agent1"
            serve_at_root: If True, forces the server to serve at root path regardless of
                http_url path component. Use this when your load balancer strips path prefixes.
                Defaults to False.
            version: The version of the agent. Defaults to "0.0.1".
            skills: The list of capabilities or functions the agent can perform.
            task_store: Custom task store implementation for managing agent tasks. If None,
                uses InMemoryTaskStore.
            queue_manager: Custom queue manager for handling message queues. If None,
                no queue management is used.
            push_config_store: Custom store for push notification configurations. If None,
                no push notification configuration is used.
            push_sender: Custom push notification sender implementation. If None,
                no push notifications are sent.
        """
        self.host = host
        self.port = port
        self.version = version

        if http_url:
            # Parse the provided URL to extract components for mounting
            self.public_base_url, self.mount_path = self._parse_public_url(http_url)
            self.http_url = http_url.rstrip("/") + "/"

            # Override mount path if serve_at_root is requested
            if serve_at_root:
                self.mount_path = ""
        else:
            # Fall back to constructing the URL from host and port
            self.public_base_url = f"http://{host}:{port}"
            self.http_url = f"{self.public_base_url}/"
            self.mount_path = ""

        self.strands_agent = agent
        self.name = self.strands_agent.name
        self.description = self.strands_agent.description
        self.capabilities = AgentCapabilities(streaming=True)
        self.request_handler = DefaultRequestHandler(
            agent_executor=StrandsA2AExecutor(self.strands_agent),
            task_store=task_store or InMemoryTaskStore(),
            queue_manager=queue_manager,
            push_config_store=push_config_store,
            push_sender=push_sender,
        )
        self._agent_skills = skills
        logger.info("Strands' integration with A2A is experimental. Be aware of frequent breaking changes.")

    def _parse_public_url(self, url: str) -> tuple[str, str]:
        """Parse the public URL into base URL and mount path components.

        Args:
            url: The full public URL (e.g., "http://my-alb.amazonaws.com/agent1")

        Returns:
            tuple: (base_url, mount_path) where base_url is the scheme+netloc
                  and mount_path is the path component

        Example:
            _parse_public_url("http://my-alb.amazonaws.com/agent1")
            Returns: ("http://my-alb.amazonaws.com", "/agent1")
        """
        parsed = urlparse(url.rstrip("/"))
        base_url = f"{parsed.scheme}://{parsed.netloc}"
        mount_path = parsed.path if parsed.path != "/" else ""
        return base_url, mount_path

    @property
    def public_agent_card(self) -> AgentCard:
        """Get the public AgentCard for this agent.

        The AgentCard contains metadata about the agent, including its name,
        description, URL, version, skills, and capabilities. This information
        is used by other agents and systems to discover and interact with this agent.

        Returns:
            AgentCard: The public agent card containing metadata about this agent.

        Raises:
            ValueError: If name or description is None or empty.
        """
        if not self.name:
            raise ValueError("A2A agent name cannot be None or empty")
        if not self.description:
            raise ValueError("A2A agent description cannot be None or empty")

        return AgentCard(
            name=self.name,
            description=self.description,
            url=self.http_url,
            version=self.version,
            skills=self.agent_skills,
            default_input_modes=["text"],
            default_output_modes=["text"],
            capabilities=self.capabilities,
        )

    def _get_skills_from_tools(self) -> list[AgentSkill]:
        """Get the list of skills from Strands agent tools.

        Skills represent specific capabilities that the agent can perform.
        Strands agent tools are adapted to A2A skills.

        Returns:
            list[AgentSkill]: A list of skills this agent provides.
        """
        return [
            AgentSkill(name=config["name"], id=config["name"], description=config["description"], tags=[])
            for config in self.strands_agent.tool_registry.get_all_tools_config().values()
        ]

    @property
    def agent_skills(self) -> list[AgentSkill]:
        """Get the list of skills this agent provides."""
        return self._agent_skills if self._agent_skills is not None else self._get_skills_from_tools()

    @agent_skills.setter
    def agent_skills(self, skills: list[AgentSkill]) -> None:
        """Set the list of skills this agent provides.

        Args:
            skills: A list of AgentSkill objects to set for this agent.
        """
        self._agent_skills = skills

    def to_starlette_app(self, *, app_kwargs: dict[str, Any] | None = None) -> Starlette:
        """Create a Starlette application for serving this agent via HTTP.

        Automatically handles path-based mounting if a mount path was derived
        from the http_url parameter.

        Args:
            app_kwargs: Additional keyword arguments to pass to the Starlette constructor.

        Returns:
            Starlette: A Starlette application configured to serve this agent.
        """
        a2a_app = A2AStarletteApplication(agent_card=self.public_agent_card, http_handler=self.request_handler).build(
            **app_kwargs or {}
        )

        if self.mount_path:
            # Create parent app and mount the A2A app at the specified path
            parent_app = Starlette()
            parent_app.mount(self.mount_path, a2a_app)
            logger.info("Mounting A2A server at path: %s", self.mount_path)
            return parent_app

        return a2a_app

    def to_fastapi_app(self, *, app_kwargs: dict[str, Any] | None = None) -> FastAPI:
        """Create a FastAPI application for serving this agent via HTTP.

        Automatically handles path-based mounting if a mount path was derived
        from the http_url parameter.

        Args:
            app_kwargs: Additional keyword arguments to pass to the FastAPI constructor.

        Returns:
            FastAPI: A FastAPI application configured to serve this agent.
        """
        a2a_app = A2AFastAPIApplication(agent_card=self.public_agent_card, http_handler=self.request_handler).build(
            **app_kwargs or {}
        )

        if self.mount_path:
            # Create parent app and mount the A2A app at the specified path
            parent_app = FastAPI()
            parent_app.mount(self.mount_path, a2a_app)
            logger.info("Mounting A2A server at path: %s", self.mount_path)
            return parent_app

        return a2a_app

    def serve(
        self,
        app_type: Literal["fastapi", "starlette"] = "starlette",
        *,
        host: str | None = None,
        port: int | None = None,
        **kwargs: Any,
    ) -> None:
        """Start the A2A server with the specified application type.

        This method starts an HTTP server that exposes the agent via the A2A protocol.
        The server can be implemented using either FastAPI or Starlette, depending on
        the specified app_type.

        Args:
            app_type: The type of application to serve, either "fastapi" or "starlette".
                Defaults to "starlette".
            host: The host address to bind the server to. Defaults to "0.0.0.0".
            port: The port number to bind the server to. Defaults to 9000.
            **kwargs: Additional keyword arguments to pass to uvicorn.run.
        """
        try:
            logger.info("Starting Strands A2A server...")
            if app_type == "fastapi":
                uvicorn.run(self.to_fastapi_app(), host=host or self.host, port=port or self.port, **kwargs)
            else:
                uvicorn.run(self.to_starlette_app(), host=host or self.host, port=port or self.port, **kwargs)
        except KeyboardInterrupt:
            logger.warning("Strands A2A server shutdown requested (KeyboardInterrupt).")
        except Exception:
            logger.exception("Strands A2A server encountered exception.")
        finally:
            logger.info("Strands A2A server has shutdown.")

agent_skills property writable

Get the list of skills this agent provides.

public_agent_card property

Get the public AgentCard for this agent.

The AgentCard contains metadata about the agent, including its name, description, URL, version, skills, and capabilities. This information is used by other agents and systems to discover and interact with this agent.

Returns:

Name Type Description
AgentCard AgentCard

The public agent card containing metadata about this agent.

Raises:

Type Description
ValueError

If name or description is None or empty.

__init__(agent, *, host='127.0.0.1', port=9000, http_url=None, serve_at_root=False, version='0.0.1', skills=None, task_store=None, queue_manager=None, push_config_store=None, push_sender=None)

Initialize an A2A-compatible server from a Strands agent.

Parameters:

Name Type Description Default
agent Agent

The Strands Agent to wrap with A2A compatibility.

 required
host str

The hostname or IP address to bind the A2A server to. Defaults to "127.0.0.1".

 '127.0.0.1'
port int

The port to bind the A2A server to. Defaults to 9000.

 9000
http_url str | None

The public HTTP URL where this agent will be accessible. If provided, this overrides the generated URL from host/port and enables automatic path-based mounting for load balancer scenarios. Example: "http://my-alb.amazonaws.com/agent1"

 None
serve_at_root bool

If True, forces the server to serve at root path regardless of http_url path component. Use this when your load balancer strips path prefixes. Defaults to False.

 False
version str

The version of the agent. Defaults to "0.0.1".

 '0.0.1'
skills list[AgentSkill] | None

The list of capabilities or functions the agent can perform.

 None
task_store TaskStore | None

Custom task store implementation for managing agent tasks. If None, uses InMemoryTaskStore.

 None
queue_manager QueueManager | None

Custom queue manager for handling message queues. If None, no queue management is used.

 None
push_config_store PushNotificationConfigStore | None

Custom store for push notification configurations. If None, no push notification configuration is used.

 None
push_sender PushNotificationSender | None

Custom push notification sender implementation. If None, no push notifications are sent.

 None
Source code in strands/multiagent/a2a/server.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def __init__(
    self,
    agent: SAAgent,
    *,
    # AgentCard
    host: str = "127.0.0.1",
    port: int = 9000,
    http_url: str | None = None,
    serve_at_root: bool = False,
    version: str = "0.0.1",
    skills: list[AgentSkill] | None = None,
    # RequestHandler
    task_store: TaskStore | None = None,
    queue_manager: QueueManager | None = None,
    push_config_store: PushNotificationConfigStore | None = None,
    push_sender: PushNotificationSender | None = None,
):
    """Initialize an A2A-compatible server from a Strands agent.

    Args:
        agent: The Strands Agent to wrap with A2A compatibility.
        host: The hostname or IP address to bind the A2A server to. Defaults to "127.0.0.1".
        port: The port to bind the A2A server to. Defaults to 9000.
        http_url: The public HTTP URL where this agent will be accessible. If provided,
            this overrides the generated URL from host/port and enables automatic
            path-based mounting for load balancer scenarios.
            Example: "http://my-alb.amazonaws.com/agent1"
        serve_at_root: If True, forces the server to serve at root path regardless of
            http_url path component. Use this when your load balancer strips path prefixes.
            Defaults to False.
        version: The version of the agent. Defaults to "0.0.1".
        skills: The list of capabilities or functions the agent can perform.
        task_store: Custom task store implementation for managing agent tasks. If None,
            uses InMemoryTaskStore.
        queue_manager: Custom queue manager for handling message queues. If None,
            no queue management is used.
        push_config_store: Custom store for push notification configurations. If None,
            no push notification configuration is used.
        push_sender: Custom push notification sender implementation. If None,
            no push notifications are sent.
    """
    self.host = host
    self.port = port
    self.version = version

    if http_url:
        # Parse the provided URL to extract components for mounting
        self.public_base_url, self.mount_path = self._parse_public_url(http_url)
        self.http_url = http_url.rstrip("/") + "/"

        # Override mount path if serve_at_root is requested
        if serve_at_root:
            self.mount_path = ""
    else:
        # Fall back to constructing the URL from host and port
        self.public_base_url = f"http://{host}:{port}"
        self.http_url = f"{self.public_base_url}/"
        self.mount_path = ""

    self.strands_agent = agent
    self.name = self.strands_agent.name
    self.description = self.strands_agent.description
    self.capabilities = AgentCapabilities(streaming=True)
    self.request_handler = DefaultRequestHandler(
        agent_executor=StrandsA2AExecutor(self.strands_agent),
        task_store=task_store or InMemoryTaskStore(),
        queue_manager=queue_manager,
        push_config_store=push_config_store,
        push_sender=push_sender,
    )
    self._agent_skills = skills
    logger.info("Strands' integration with A2A is experimental. Be aware of frequent breaking changes.")

serve(app_type='starlette', *, host=None, port=None, **kwargs)

Start the A2A server with the specified application type.

This method starts an HTTP server that exposes the agent via the A2A protocol. The server can be implemented using either FastAPI or Starlette, depending on the specified app_type.

Parameters:

Name Type Description Default
app_type Literal['fastapi', 'starlette']

The type of application to serve, either "fastapi" or "starlette". Defaults to "starlette".

 'starlette'
host str | None

The host address to bind the server to. Defaults to "0.0.0.0".

 None
port int | None

The port number to bind the server to. Defaults to 9000.

 None
**kwargs Any

Additional keyword arguments to pass to uvicorn.run.

 {}
Source code in strands/multiagent/a2a/server.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def serve(
    self,
    app_type: Literal["fastapi", "starlette"] = "starlette",
    *,
    host: str | None = None,
    port: int | None = None,
    **kwargs: Any,
) -> None:
    """Start the A2A server with the specified application type.

    This method starts an HTTP server that exposes the agent via the A2A protocol.
    The server can be implemented using either FastAPI or Starlette, depending on
    the specified app_type.

    Args:
        app_type: The type of application to serve, either "fastapi" or "starlette".
            Defaults to "starlette".
        host: The host address to bind the server to. Defaults to "0.0.0.0".
        port: The port number to bind the server to. Defaults to 9000.
        **kwargs: Additional keyword arguments to pass to uvicorn.run.
    """
    try:
        logger.info("Starting Strands A2A server...")
        if app_type == "fastapi":
            uvicorn.run(self.to_fastapi_app(), host=host or self.host, port=port or self.port, **kwargs)
        else:
            uvicorn.run(self.to_starlette_app(), host=host or self.host, port=port or self.port, **kwargs)
    except KeyboardInterrupt:
        logger.warning("Strands A2A server shutdown requested (KeyboardInterrupt).")
    except Exception:
        logger.exception("Strands A2A server encountered exception.")
    finally:
        logger.info("Strands A2A server has shutdown.")

to_fastapi_app(*, app_kwargs=None)

Create a FastAPI application for serving this agent via HTTP.

Automatically handles path-based mounting if a mount path was derived from the http_url parameter.

Parameters:

Name Type Description Default
app_kwargs dict[str, Any] | None

Additional keyword arguments to pass to the FastAPI constructor.

 None

Returns:

Name Type Description
FastAPI FastAPI

A FastAPI application configured to serve this agent.
Source code in strands/multiagent/a2a/server.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def to_fastapi_app(self, *, app_kwargs: dict[str, Any] | None = None) -> FastAPI:
    """Create a FastAPI application for serving this agent via HTTP.

    Automatically handles path-based mounting if a mount path was derived
    from the http_url parameter.

    Args:
        app_kwargs: Additional keyword arguments to pass to the FastAPI constructor.

    Returns:
        FastAPI: A FastAPI application configured to serve this agent.
    """
    a2a_app = A2AFastAPIApplication(agent_card=self.public_agent_card, http_handler=self.request_handler).build(
        **app_kwargs or {}
    )

    if self.mount_path:
        # Create parent app and mount the A2A app at the specified path
        parent_app = FastAPI()
        parent_app.mount(self.mount_path, a2a_app)
        logger.info("Mounting A2A server at path: %s", self.mount_path)
        return parent_app

    return a2a_app

to_starlette_app(*, app_kwargs=None)

Create a Starlette application for serving this agent via HTTP.

Automatically handles path-based mounting if a mount path was derived from the http_url parameter.

Parameters:

Name Type Description Default
app_kwargs dict[str, Any] | None

Additional keyword arguments to pass to the Starlette constructor.

 None

Returns:

Name Type Description
Starlette Starlette

A Starlette application configured to serve this agent.
Source code in strands/multiagent/a2a/server.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def to_starlette_app(self, *, app_kwargs: dict[str, Any] | None = None) -> Starlette:
    """Create a Starlette application for serving this agent via HTTP.

    Automatically handles path-based mounting if a mount path was derived
    from the http_url parameter.

    Args:
        app_kwargs: Additional keyword arguments to pass to the Starlette constructor.

    Returns:
        Starlette: A Starlette application configured to serve this agent.
    """
    a2a_app = A2AStarletteApplication(agent_card=self.public_agent_card, http_handler=self.request_handler).build(
        **app_kwargs or {}
    )

    if self.mount_path:
        # Create parent app and mount the A2A app at the specified path
        parent_app = Starlette()
        parent_app.mount(self.mount_path, a2a_app)
        logger.info("Mounting A2A server at path: %s", self.mount_path)
        return parent_app

    return a2a_app

SAAgent

Core Agent interface.

An agent orchestrates the following workflow:

  1. Receives user input
  2. Processes the input using a language model
  3. Decides whether to use tools to gather information or perform actions
  4. Executes those tools and receives results
  5. Continues reasoning with the new information
  6. Produces a final response
Source code in strands/agent/agent.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
class Agent:
    """Core Agent interface.

    An agent orchestrates the following workflow:

    1. Receives user input
    2. Processes the input using a language model
    3. Decides whether to use tools to gather information or perform actions
    4. Executes those tools and receives results
    5. Continues reasoning with the new information
    6. Produces a final response
    """

    # For backwards compatibility
    ToolCaller = _ToolCaller

    def __init__(
        self,
        model: Union[Model, str, None] = None,
        messages: Optional[Messages] = None,
        tools: Optional[list[Union[str, dict[str, str], "ToolProvider", Any]]] = None,
        system_prompt: Optional[str | list[SystemContentBlock]] = None,
        structured_output_model: Optional[Type[BaseModel]] = None,
        callback_handler: Optional[
            Union[Callable[..., Any], _DefaultCallbackHandlerSentinel]
        ] = _DEFAULT_CALLBACK_HANDLER,
        conversation_manager: Optional[ConversationManager] = None,
        record_direct_tool_call: bool = True,
        load_tools_from_directory: bool = False,
        trace_attributes: Optional[Mapping[str, AttributeValue]] = None,
        *,
        agent_id: Optional[str] = None,
        name: Optional[str] = None,
        description: Optional[str] = None,
        state: Optional[Union[AgentState, dict]] = None,
        hooks: Optional[list[HookProvider]] = None,
        session_manager: Optional[SessionManager] = None,
        tool_executor: Optional[ToolExecutor] = None,
    ):
        """Initialize the Agent with the specified configuration.

        Args:
            model: Provider for running inference or a string representing the model-id for Bedrock to use.
                Defaults to strands.models.BedrockModel if None.
            messages: List of initial messages to pre-load into the conversation.
                Defaults to an empty list if None.
            tools: List of tools to make available to the agent.
                Can be specified as:

                - String tool names (e.g., "retrieve")
                - File paths (e.g., "/path/to/tool.py")
                - Imported Python modules (e.g., from strands_tools import current_time)
                - Dictionaries with name/path keys (e.g., {"name": "tool_name", "path": "/path/to/tool.py"})
                - ToolProvider instances for managed tool collections
                - Functions decorated with `@strands.tool` decorator.

                If provided, only these tools will be available. If None, all tools will be available.
            system_prompt: System prompt to guide model behavior.
                Can be a string or a list of SystemContentBlock objects for advanced features like caching.
                If None, the model will behave according to its default settings.
            structured_output_model: Pydantic model type(s) for structured output.
                When specified, all agent calls will attempt to return structured output of this type.
                This can be overridden on the agent invocation.
                Defaults to None (no structured output).
            callback_handler: Callback for processing events as they happen during agent execution.
                If not provided (using the default), a new PrintingCallbackHandler instance is created.
                If explicitly set to None, null_callback_handler is used.
            conversation_manager: Manager for conversation history and context window.
                Defaults to strands.agent.conversation_manager.SlidingWindowConversationManager if None.
            record_direct_tool_call: Whether to record direct tool calls in message history.
                Defaults to True.
            load_tools_from_directory: Whether to load and automatically reload tools in the `./tools/` directory.
                Defaults to False.
            trace_attributes: Custom trace attributes to apply to the agent's trace span.
            agent_id: Optional ID for the agent, useful for session management and multi-agent scenarios.
                Defaults to "default".
            name: name of the Agent
                Defaults to "Strands Agents".
            description: description of what the Agent does
                Defaults to None.
            state: stateful information for the agent. Can be either an AgentState object, or a json serializable dict.
                Defaults to an empty AgentState object.
            hooks: hooks to be added to the agent hook registry
                Defaults to None.
            session_manager: Manager for handling agent sessions including conversation history and state.
                If provided, enables session-based persistence and state management.
            tool_executor: Definition of tool execution strategy (e.g., sequential, concurrent, etc.).

        Raises:
            ValueError: If agent id contains path separators.
        """
        self.model = BedrockModel() if not model else BedrockModel(model_id=model) if isinstance(model, str) else model
        self.messages = messages if messages is not None else []
        # initializing self._system_prompt for backwards compatibility
        self._system_prompt, self._system_prompt_content = self._initialize_system_prompt(system_prompt)
        self._default_structured_output_model = structured_output_model
        self.agent_id = _identifier.validate(agent_id or _DEFAULT_AGENT_ID, _identifier.Identifier.AGENT)
        self.name = name or _DEFAULT_AGENT_NAME
        self.description = description

        # If not provided, create a new PrintingCallbackHandler instance
        # If explicitly set to None, use null_callback_handler
        # Otherwise use the passed callback_handler
        self.callback_handler: Union[Callable[..., Any], PrintingCallbackHandler]
        if isinstance(callback_handler, _DefaultCallbackHandlerSentinel):
            self.callback_handler = PrintingCallbackHandler()
        elif callback_handler is None:
            self.callback_handler = null_callback_handler
        else:
            self.callback_handler = callback_handler

        self.conversation_manager = conversation_manager if conversation_manager else SlidingWindowConversationManager()

        # Process trace attributes to ensure they're of compatible types
        self.trace_attributes: dict[str, AttributeValue] = {}
        if trace_attributes:
            for k, v in trace_attributes.items():
                if isinstance(v, (str, int, float, bool)) or (
                    isinstance(v, list) and all(isinstance(x, (str, int, float, bool)) for x in v)
                ):
                    self.trace_attributes[k] = v

        self.record_direct_tool_call = record_direct_tool_call
        self.load_tools_from_directory = load_tools_from_directory

        self.tool_registry = ToolRegistry()

        # Process tool list if provided
        if tools is not None:
            self.tool_registry.process_tools(tools)

        # Initialize tools and configuration
        self.tool_registry.initialize_tools(self.load_tools_from_directory)
        if load_tools_from_directory:
            self.tool_watcher = ToolWatcher(tool_registry=self.tool_registry)

        self.event_loop_metrics = EventLoopMetrics()

        # Initialize tracer instance (no-op if not configured)
        self.tracer = get_tracer()
        self.trace_span: Optional[trace_api.Span] = None

        # Initialize agent state management
        if state is not None:
            if isinstance(state, dict):
                self.state = AgentState(state)
            elif isinstance(state, AgentState):
                self.state = state
            else:
                raise ValueError("state must be an AgentState object or a dict")
        else:
            self.state = AgentState()

        self.tool_caller = _ToolCaller(self)

        self.hooks = HookRegistry()

        self._interrupt_state = _InterruptState()

        # Initialize session management functionality
        self._session_manager = session_manager
        if self._session_manager:
            self.hooks.add_hook(self._session_manager)

        # Allow conversation_managers to subscribe to hooks
        self.hooks.add_hook(self.conversation_manager)

        self.tool_executor = tool_executor or ConcurrentToolExecutor()

        if hooks:
            for hook in hooks:
                self.hooks.add_hook(hook)
        self.hooks.invoke_callbacks(AgentInitializedEvent(agent=self))

    @property
    def system_prompt(self) -> str | None:
        """Get the system prompt as a string for backwards compatibility.

        Returns the system prompt as a concatenated string when it contains text content,
        or None if no text content is present. This maintains backwards compatibility
        with existing code that expects system_prompt to be a string.

        Returns:
            The system prompt as a string, or None if no text content exists.
        """
        return self._system_prompt

    @system_prompt.setter
    def system_prompt(self, value: str | list[SystemContentBlock] | None) -> None:
        """Set the system prompt and update internal content representation.

        Accepts either a string or list of SystemContentBlock objects.
        When set, both the backwards-compatible string representation and the internal
        content block representation are updated to maintain consistency.

        Args:
            value: System prompt as string, list of SystemContentBlock objects, or None.
                  - str: Simple text prompt (most common use case)
                  - list[SystemContentBlock]: Content blocks with features like caching
                  - None: Clear the system prompt
        """
        self._system_prompt, self._system_prompt_content = self._initialize_system_prompt(value)

    @property
    def tool(self) -> _ToolCaller:
        """Call tool as a function.

        Returns:
            Tool caller through which user can invoke tool as a function.

        Example:
            ```
            agent = Agent(tools=[calculator])
            agent.tool.calculator(...)
            ```
        """
        return self.tool_caller

    @property
    def tool_names(self) -> list[str]:
        """Get a list of all registered tool names.

        Returns:
            Names of all tools available to this agent.
        """
        all_tools = self.tool_registry.get_all_tools_config()
        return list(all_tools.keys())

    def __call__(
        self,
        prompt: AgentInput = None,
        *,
        invocation_state: dict[str, Any] | None = None,
        structured_output_model: Type[BaseModel] | None = None,
        **kwargs: Any,
    ) -> AgentResult:
        """Process a natural language prompt through the agent's event loop.

        This method implements the conversational interface with multiple input patterns:
        - String input: `agent("hello!")`
        - ContentBlock list: `agent([{"text": "hello"}, {"image": {...}}])`
        - Message list: `agent([{"role": "user", "content": [{"text": "hello"}]}])`
        - No input: `agent()` - uses existing conversation history

        Args:
            prompt: User input in various formats:
                - str: Simple text input
                - list[ContentBlock]: Multi-modal content blocks
                - list[Message]: Complete messages with roles
                - None: Use existing conversation history
            invocation_state: Additional parameters to pass through the event loop.
            structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
            **kwargs: Additional parameters to pass through the event loop.[Deprecating]

        Returns:
            Result object containing:

                - stop_reason: Why the event loop stopped (e.g., "end_turn", "max_tokens")
                - message: The final message from the model
                - metrics: Performance metrics from the event loop
                - state: The final state of the event loop
                - structured_output: Parsed structured output when structured_output_model was specified
        """
        return run_async(
            lambda: self.invoke_async(
                prompt, invocation_state=invocation_state, structured_output_model=structured_output_model, **kwargs
            )
        )

    async def invoke_async(
        self,
        prompt: AgentInput = None,
        *,
        invocation_state: dict[str, Any] | None = None,
        structured_output_model: Type[BaseModel] | None = None,
        **kwargs: Any,
    ) -> AgentResult:
        """Process a natural language prompt through the agent's event loop.

        This method implements the conversational interface with multiple input patterns:
        - String input: Simple text input
        - ContentBlock list: Multi-modal content blocks
        - Message list: Complete messages with roles
        - No input: Use existing conversation history

        Args:
            prompt: User input in various formats:
                - str: Simple text input
                - list[ContentBlock]: Multi-modal content blocks
                - list[Message]: Complete messages with roles
                - None: Use existing conversation history
            invocation_state: Additional parameters to pass through the event loop.
            structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
            **kwargs: Additional parameters to pass through the event loop.[Deprecating]

        Returns:
            Result: object containing:

                - stop_reason: Why the event loop stopped (e.g., "end_turn", "max_tokens")
                - message: The final message from the model
                - metrics: Performance metrics from the event loop
                - state: The final state of the event loop
        """
        events = self.stream_async(
            prompt, invocation_state=invocation_state, structured_output_model=structured_output_model, **kwargs
        )
        async for event in events:
            _ = event

        return cast(AgentResult, event["result"])

    def structured_output(self, output_model: Type[T], prompt: AgentInput = None) -> T:
        """This method allows you to get structured output from the agent.

        If you pass in a prompt, it will be used temporarily without adding it to the conversation history.
        If you don't pass in a prompt, it will use only the existing conversation history to respond.

        For smaller models, you may want to use the optional prompt to add additional instructions to explicitly
        instruct the model to output the structured data.

        Args:
            output_model: The output model (a JSON schema written as a Pydantic BaseModel)
                that the agent will use when responding.
            prompt: The prompt to use for the agent in various formats:
                - str: Simple text input
                - list[ContentBlock]: Multi-modal content blocks
                - list[Message]: Complete messages with roles
                - None: Use existing conversation history

        Raises:
            ValueError: If no conversation history or prompt is provided.
        """
        warnings.warn(
            "Agent.structured_output method is deprecated."
            " You should pass in `structured_output_model` directly into the agent invocation."
            " see: https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/structured-output/",
            category=DeprecationWarning,
            stacklevel=2,
        )

        return run_async(lambda: self.structured_output_async(output_model, prompt))

    async def structured_output_async(self, output_model: Type[T], prompt: AgentInput = None) -> T:
        """This method allows you to get structured output from the agent.

        If you pass in a prompt, it will be used temporarily without adding it to the conversation history.
        If you don't pass in a prompt, it will use only the existing conversation history to respond.

        For smaller models, you may want to use the optional prompt to add additional instructions to explicitly
        instruct the model to output the structured data.

        Args:
            output_model: The output model (a JSON schema written as a Pydantic BaseModel)
                that the agent will use when responding.
            prompt: The prompt to use for the agent (will not be added to conversation history).

        Raises:
            ValueError: If no conversation history or prompt is provided.
        -
        """
        if self._interrupt_state.activated:
            raise RuntimeError("cannot call structured output during interrupt")

        warnings.warn(
            "Agent.structured_output_async method is deprecated."
            " You should pass in `structured_output_model` directly into the agent invocation."
            " see: https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/structured-output/",
            category=DeprecationWarning,
            stacklevel=2,
        )
        await self.hooks.invoke_callbacks_async(BeforeInvocationEvent(agent=self))
        with self.tracer.tracer.start_as_current_span(
            "execute_structured_output", kind=trace_api.SpanKind.CLIENT
        ) as structured_output_span:
            try:
                if not self.messages and not prompt:
                    raise ValueError("No conversation history or prompt provided")

                temp_messages: Messages = self.messages + await self._convert_prompt_to_messages(prompt)

                structured_output_span.set_attributes(
                    {
                        "gen_ai.system": "strands-agents",
                        "gen_ai.agent.name": self.name,
                        "gen_ai.agent.id": self.agent_id,
                        "gen_ai.operation.name": "execute_structured_output",
                    }
                )
                if self.system_prompt:
                    structured_output_span.add_event(
                        "gen_ai.system.message",
                        attributes={"role": "system", "content": serialize([{"text": self.system_prompt}])},
                    )
                for message in temp_messages:
                    structured_output_span.add_event(
                        f"gen_ai.{message['role']}.message",
                        attributes={"role": message["role"], "content": serialize(message["content"])},
                    )
                events = self.model.structured_output(output_model, temp_messages, system_prompt=self.system_prompt)
                async for event in events:
                    if isinstance(event, TypedEvent):
                        event.prepare(invocation_state={})
                        if event.is_callback_event:
                            self.callback_handler(**event.as_dict())

                structured_output_span.add_event(
                    "gen_ai.choice", attributes={"message": serialize(event["output"].model_dump())}
                )
                return event["output"]

            finally:
                await self.hooks.invoke_callbacks_async(AfterInvocationEvent(agent=self))

    def cleanup(self) -> None:
        """Clean up resources used by the agent.

        This method cleans up all tool providers that require explicit cleanup,
        such as MCP clients. It should be called when the agent is no longer needed
        to ensure proper resource cleanup.

        Note: This method uses a "belt and braces" approach with automatic cleanup
        through finalizers as a fallback, but explicit cleanup is recommended.
        """
        self.tool_registry.cleanup()

    def __del__(self) -> None:
        """Clean up resources when agent is garbage collected."""
        # __del__ is called even when an exception is thrown in the constructor,
        # so there is no guarantee tool_registry was set..
        if hasattr(self, "tool_registry"):
            self.tool_registry.cleanup()

    async def stream_async(
        self,
        prompt: AgentInput = None,
        *,
        invocation_state: dict[str, Any] | None = None,
        structured_output_model: Type[BaseModel] | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[Any]:
        """Process a natural language prompt and yield events as an async iterator.

        This method provides an asynchronous interface for streaming agent events with multiple input patterns:
        - String input: Simple text input
        - ContentBlock list: Multi-modal content blocks
        - Message list: Complete messages with roles
        - No input: Use existing conversation history

        Args:
            prompt: User input in various formats:
                - str: Simple text input
                - list[ContentBlock]: Multi-modal content blocks
                - list[Message]: Complete messages with roles
                - None: Use existing conversation history
            invocation_state: Additional parameters to pass through the event loop.
            structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
            **kwargs: Additional parameters to pass to the event loop.[Deprecating]

        Yields:
            An async iterator that yields events. Each event is a dictionary containing
                information about the current state of processing, such as:

                - data: Text content being generated
                - complete: Whether this is the final chunk
                - current_tool_use: Information about tools being executed
                - And other event data provided by the callback handler

        Raises:
            Exception: Any exceptions from the agent invocation will be propagated to the caller.

        Example:
            ```python
            async for event in agent.stream_async("Analyze this data"):
                if "data" in event:
                    yield event["data"]
            ```
        """
        self._interrupt_state.resume(prompt)

        self.event_loop_metrics.reset_usage_metrics()

        merged_state = {}
        if kwargs:
            warnings.warn("`**kwargs` parameter is deprecating, use `invocation_state` instead.", stacklevel=2)
            merged_state.update(kwargs)
            if invocation_state is not None:
                merged_state["invocation_state"] = invocation_state
        else:
            if invocation_state is not None:
                merged_state = invocation_state

        callback_handler = self.callback_handler
        if kwargs:
            callback_handler = kwargs.get("callback_handler", self.callback_handler)

        # Process input and get message to add (if any)
        messages = await self._convert_prompt_to_messages(prompt)

        self.trace_span = self._start_agent_trace_span(messages)

        with trace_api.use_span(self.trace_span):
            try:
                events = self._run_loop(messages, merged_state, structured_output_model)

                async for event in events:
                    event.prepare(invocation_state=merged_state)

                    if event.is_callback_event:
                        as_dict = event.as_dict()
                        callback_handler(**as_dict)
                        yield as_dict

                result = AgentResult(*event["stop"])
                callback_handler(result=result)
                yield AgentResultEvent(result=result).as_dict()

                self._end_agent_trace_span(response=result)

            except Exception as e:
                self._end_agent_trace_span(error=e)
                raise

    async def _run_loop(
        self,
        messages: Messages,
        invocation_state: dict[str, Any],
        structured_output_model: Type[BaseModel] | None = None,
    ) -> AsyncGenerator[TypedEvent, None]:
        """Execute the agent's event loop with the given message and parameters.

        Args:
            messages: The input messages to add to the conversation.
            invocation_state: Additional parameters to pass to the event loop.
            structured_output_model: Optional Pydantic model type for structured output.

        Yields:
            Events from the event loop cycle.
        """
        await self.hooks.invoke_callbacks_async(BeforeInvocationEvent(agent=self))

        agent_result: AgentResult | None = None
        try:
            yield InitEventLoopEvent()

            await self._append_messages(*messages)

            structured_output_context = StructuredOutputContext(
                structured_output_model or self._default_structured_output_model
            )

            # Execute the event loop cycle with retry logic for context limits
            events = self._execute_event_loop_cycle(invocation_state, structured_output_context)
            async for event in events:
                # Signal from the model provider that the message sent by the user should be redacted,
                # likely due to a guardrail.
                if (
                    isinstance(event, ModelStreamChunkEvent)
                    and event.chunk
                    and event.chunk.get("redactContent")
                    and event.chunk["redactContent"].get("redactUserContentMessage")
                ):
                    self.messages[-1]["content"] = self._redact_user_content(
                        self.messages[-1]["content"], str(event.chunk["redactContent"]["redactUserContentMessage"])
                    )
                    if self._session_manager:
                        self._session_manager.redact_latest_message(self.messages[-1], self)
                yield event

            # Capture the result from the final event if available
            if isinstance(event, EventLoopStopEvent):
                agent_result = AgentResult(*event["stop"])

        finally:
            self.conversation_manager.apply_management(self)
            await self.hooks.invoke_callbacks_async(AfterInvocationEvent(agent=self, result=agent_result))

    async def _execute_event_loop_cycle(
        self, invocation_state: dict[str, Any], structured_output_context: StructuredOutputContext | None = None
    ) -> AsyncGenerator[TypedEvent, None]:
        """Execute the event loop cycle with retry logic for context window limits.

        This internal method handles the execution of the event loop cycle and implements
        retry logic for handling context window overflow exceptions by reducing the
        conversation context and retrying.

        Args:
            invocation_state: Additional parameters to pass to the event loop.
            structured_output_context: Optional structured output context for this invocation.

        Yields:
            Events of the loop cycle.
        """
        # Add `Agent` to invocation_state to keep backwards-compatibility
        invocation_state["agent"] = self

        if structured_output_context:
            structured_output_context.register_tool(self.tool_registry)

        try:
            events = event_loop_cycle(
                agent=self,
                invocation_state=invocation_state,
                structured_output_context=structured_output_context,
            )
            async for event in events:
                yield event

        except ContextWindowOverflowException as e:
            # Try reducing the context size and retrying
            self.conversation_manager.reduce_context(self, e=e)

            # Sync agent after reduce_context to keep conversation_manager_state up to date in the session
            if self._session_manager:
                self._session_manager.sync_agent(self)

            events = self._execute_event_loop_cycle(invocation_state, structured_output_context)
            async for event in events:
                yield event

        finally:
            if structured_output_context:
                structured_output_context.cleanup(self.tool_registry)

    async def _convert_prompt_to_messages(self, prompt: AgentInput) -> Messages:
        if self._interrupt_state.activated:
            return []

        messages: Messages | None = None
        if prompt is not None:
            # Check if the latest message is toolUse
            if len(self.messages) > 0 and any("toolUse" in content for content in self.messages[-1]["content"]):
                # Add toolResult message after to have a valid conversation
                logger.info(
                    "Agents latest message is toolUse, appending a toolResult message to have valid conversation."
                )
                tool_use_ids = [
                    content["toolUse"]["toolUseId"] for content in self.messages[-1]["content"] if "toolUse" in content
                ]
                await self._append_messages(
                    {
                        "role": "user",
                        "content": generate_missing_tool_result_content(tool_use_ids),
                    }
                )
            if isinstance(prompt, str):
                # String input - convert to user message
                messages = [{"role": "user", "content": [{"text": prompt}]}]
            elif isinstance(prompt, list):
                if len(prompt) == 0:
                    # Empty list
                    messages = []
                # Check if all item in input list are dictionaries
                elif all(isinstance(item, dict) for item in prompt):
                    # Check if all items are messages
                    if all(all(key in item for key in Message.__annotations__.keys()) for item in prompt):
                        # Messages input - add all messages to conversation
                        messages = cast(Messages, prompt)

                    # Check if all items are content blocks
                    elif all(any(key in ContentBlock.__annotations__.keys() for key in item) for item in prompt):
                        # Treat as List[ContentBlock] input - convert to user message
                        # This allows invalid structures to be passed through to the model
                        messages = [{"role": "user", "content": cast(list[ContentBlock], prompt)}]
        else:
            messages = []
        if messages is None:
            raise ValueError("Input prompt must be of type: `str | list[Contentblock] | Messages | None`.")
        return messages

    def _start_agent_trace_span(self, messages: Messages) -> trace_api.Span:
        """Starts a trace span for the agent.

        Args:
            messages: The input messages.
        """
        model_id = self.model.config.get("model_id") if hasattr(self.model, "config") else None
        return self.tracer.start_agent_span(
            messages=messages,
            agent_name=self.name,
            model_id=model_id,
            tools=self.tool_names,
            system_prompt=self.system_prompt,
            custom_trace_attributes=self.trace_attributes,
            tools_config=self.tool_registry.get_all_tools_config(),
        )

    def _end_agent_trace_span(
        self,
        response: Optional[AgentResult] = None,
        error: Optional[Exception] = None,
    ) -> None:
        """Ends a trace span for the agent.

        Args:
            span: The span to end.
            response: Response to record as a trace attribute.
            error: Error to record as a trace attribute.
        """
        if self.trace_span:
            trace_attributes: dict[str, Any] = {
                "span": self.trace_span,
            }

            if response:
                trace_attributes["response"] = response
            if error:
                trace_attributes["error"] = error

            self.tracer.end_agent_span(**trace_attributes)

    def _initialize_system_prompt(
        self, system_prompt: str | list[SystemContentBlock] | None
    ) -> tuple[str | None, list[SystemContentBlock] | None]:
        """Initialize system prompt fields from constructor input.

        Maintains backwards compatibility by keeping system_prompt as str when string input
        provided, avoiding breaking existing consumers.

        Maps system_prompt input to both string and content block representations:
        - If string: system_prompt=string, _system_prompt_content=[{text: string}]
        - If list with text elements: system_prompt=concatenated_text, _system_prompt_content=list
        - If list without text elements: system_prompt=None, _system_prompt_content=list
        - If None: system_prompt=None, _system_prompt_content=None
        """
        if isinstance(system_prompt, str):
            return system_prompt, [{"text": system_prompt}]
        elif isinstance(system_prompt, list):
            # Concatenate all text elements for backwards compatibility, None if no text found
            text_parts = [block["text"] for block in system_prompt if "text" in block]
            system_prompt_str = "\n".join(text_parts) if text_parts else None
            return system_prompt_str, system_prompt
        else:
            return None, None

    async def _append_messages(self, *messages: Message) -> None:
        """Appends messages to history and invoke the callbacks for the MessageAddedEvent."""
        for message in messages:
            self.messages.append(message)
            await self.hooks.invoke_callbacks_async(MessageAddedEvent(agent=self, message=message))

    def _redact_user_content(self, content: list[ContentBlock], redact_message: str) -> list[ContentBlock]:
        """Redact user content preserving toolResult blocks.

        Args:
            content: content blocks to be redacted
            redact_message: redact message to be replaced

        Returns:
            Redacted content, as follows:
            - if the message contains at least a toolResult block,
                all toolResult blocks(s) are kept, redacting only the result content;
            - otherwise, the entire content of the message is replaced
                with a single text block with the redact message.
        """
        redacted_content = []
        for block in content:
            if "toolResult" in block:
                block["toolResult"]["content"] = [{"text": redact_message}]
                redacted_content.append(block)

        if not redacted_content:
            # Text content is added only if no toolResult blocks were found
            redacted_content = [{"text": redact_message}]

        return redacted_content

system_prompt property writable

Get the system prompt as a string for backwards compatibility.

Returns the system prompt as a concatenated string when it contains text content, or None if no text content is present. This maintains backwards compatibility with existing code that expects system_prompt to be a string.

Returns:

Type Description
str | None

The system prompt as a string, or None if no text content exists.

tool property

Call tool as a function.

Returns:

Type Description
_ToolCaller

Tool caller through which user can invoke tool as a function.
Example 
agent = Agent(tools=[calculator])
agent.tool.calculator(...)

tool_names property

Get a list of all registered tool names.

Returns:

Type Description
list[str]

Names of all tools available to this agent.

__call__(prompt=None, *, invocation_state=None, structured_output_model=None, **kwargs)

Process a natural language prompt through the agent's event loop.

This method implements the conversational interface with multiple input patterns: - String input: agent("hello!") - ContentBlock list: agent([{"text": "hello"}, {"image": {...}}]) - Message list: agent([{"role": "user", "content": [{"text": "hello"}]}]) - No input: agent() - uses existing conversation history

Parameters:

Name Type Description Default
prompt AgentInput

User input in various formats: - str: Simple text input - list[ContentBlock]: Multi-modal content blocks - list[Message]: Complete messages with roles - None: Use existing conversation history

 None
invocation_state dict[str, Any] | None

Additional parameters to pass through the event loop.

 None
structured_output_model Type[BaseModel] | None

Pydantic model type(s) for structured output (overrides agent default).

 None
**kwargs Any

Additional parameters to pass through the event loop.[Deprecating]

 {}

Returns:

Type Description
AgentResult

Result object containing:

  • stop_reason: Why the event loop stopped (e.g., "end_turn", "max_tokens")
  • message: The final message from the model
  • metrics: Performance metrics from the event loop
  • state: The final state of the event loop
  • structured_output: Parsed structured output when structured_output_model was specified
Source code in strands/agent/agent.py 
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
def __call__(
    self,
    prompt: AgentInput = None,
    *,
    invocation_state: dict[str, Any] | None = None,
    structured_output_model: Type[BaseModel] | None = None,
    **kwargs: Any,
) -> AgentResult:
    """Process a natural language prompt through the agent's event loop.

    This method implements the conversational interface with multiple input patterns:
    - String input: `agent("hello!")`
    - ContentBlock list: `agent([{"text": "hello"}, {"image": {...}}])`
    - Message list: `agent([{"role": "user", "content": [{"text": "hello"}]}])`
    - No input: `agent()` - uses existing conversation history

    Args:
        prompt: User input in various formats:
            - str: Simple text input
            - list[ContentBlock]: Multi-modal content blocks
            - list[Message]: Complete messages with roles
            - None: Use existing conversation history
        invocation_state: Additional parameters to pass through the event loop.
        structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
        **kwargs: Additional parameters to pass through the event loop.[Deprecating]

    Returns:
        Result object containing:

            - stop_reason: Why the event loop stopped (e.g., "end_turn", "max_tokens")
            - message: The final message from the model
            - metrics: Performance metrics from the event loop
            - state: The final state of the event loop
            - structured_output: Parsed structured output when structured_output_model was specified
    """
    return run_async(
        lambda: self.invoke_async(
            prompt, invocation_state=invocation_state, structured_output_model=structured_output_model, **kwargs
        )
    )

__del__()

Clean up resources when agent is garbage collected.

Source code in strands/agent/agent.py 
514
515
516
517
518
519
def __del__(self) -> None:
    """Clean up resources when agent is garbage collected."""
    # __del__ is called even when an exception is thrown in the constructor,
    # so there is no guarantee tool_registry was set..
    if hasattr(self, "tool_registry"):
        self.tool_registry.cleanup()

__init__(model=None, messages=None, tools=None, system_prompt=None, structured_output_model=None, callback_handler=_DEFAULT_CALLBACK_HANDLER, conversation_manager=None, record_direct_tool_call=True, load_tools_from_directory=False, trace_attributes=None, *, agent_id=None, name=None, description=None, state=None, hooks=None, session_manager=None, tool_executor=None)

Initialize the Agent with the specified configuration.

Parameters:

Name Type Description Default
model Union[Model, str, None]

Provider for running inference or a string representing the model-id for Bedrock to use. Defaults to strands.models.BedrockModel if None.

 None
messages Optional[Messages]

List of initial messages to pre-load into the conversation. Defaults to an empty list if None.

 None
tools Optional[list[Union[str, dict[str, str], ToolProvider, Any]]]

List of tools to make available to the agent. Can be specified as:

  • String tool names (e.g., "retrieve")
  • File paths (e.g., "/path/to/tool.py")
  • Imported Python modules (e.g., from strands_tools import current_time)
  • Dictionaries with name/path keys (e.g., {"name": "tool_name", "path": "/path/to/tool.py"})
  • ToolProvider instances for managed tool collections
  • Functions decorated with @strands.tool decorator.

If provided, only these tools will be available. If None, all tools will be available.

 None
system_prompt Optional[str | list[SystemContentBlock]]

System prompt to guide model behavior. Can be a string or a list of SystemContentBlock objects for advanced features like caching. If None, the model will behave according to its default settings.

 None
structured_output_model Optional[Type[BaseModel]]

Pydantic model type(s) for structured output. When specified, all agent calls will attempt to return structured output of this type. This can be overridden on the agent invocation. Defaults to None (no structured output).

 None
callback_handler Optional[Union[Callable[..., Any], _DefaultCallbackHandlerSentinel]]

Callback for processing events as they happen during agent execution. If not provided (using the default), a new PrintingCallbackHandler instance is created. If explicitly set to None, null_callback_handler is used.

 _DEFAULT_CALLBACK_HANDLER
conversation_manager Optional[ConversationManager]

Manager for conversation history and context window. Defaults to strands.agent.conversation_manager.SlidingWindowConversationManager if None.

 None
record_direct_tool_call bool

Whether to record direct tool calls in message history. Defaults to True.

 True
load_tools_from_directory bool

Whether to load and automatically reload tools in the ./tools/ directory. Defaults to False.

 False
trace_attributes Optional[Mapping[str, AttributeValue]]

Custom trace attributes to apply to the agent's trace span.

 None
agent_id Optional[str]

Optional ID for the agent, useful for session management and multi-agent scenarios. Defaults to "default".

 None
name Optional[str]

name of the Agent Defaults to "Strands Agents".

 None
description Optional[str]

description of what the Agent does Defaults to None.

 None
state Optional[Union[AgentState, dict]]

stateful information for the agent. Can be either an AgentState object, or a json serializable dict. Defaults to an empty AgentState object.

 None
hooks Optional[list[HookProvider]]

hooks to be added to the agent hook registry Defaults to None.

 None
session_manager Optional[SessionManager]

Manager for handling agent sessions including conversation history and state. If provided, enables session-based persistence and state management.

 None
tool_executor Optional[ToolExecutor]

Definition of tool execution strategy (e.g., sequential, concurrent, etc.).

 None

Raises:

Type Description
ValueError

If agent id contains path separators.
Source code in strands/agent/agent.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def __init__(
    self,
    model: Union[Model, str, None] = None,
    messages: Optional[Messages] = None,
    tools: Optional[list[Union[str, dict[str, str], "ToolProvider", Any]]] = None,
    system_prompt: Optional[str | list[SystemContentBlock]] = None,
    structured_output_model: Optional[Type[BaseModel]] = None,
    callback_handler: Optional[
        Union[Callable[..., Any], _DefaultCallbackHandlerSentinel]
    ] = _DEFAULT_CALLBACK_HANDLER,
    conversation_manager: Optional[ConversationManager] = None,
    record_direct_tool_call: bool = True,
    load_tools_from_directory: bool = False,
    trace_attributes: Optional[Mapping[str, AttributeValue]] = None,
    *,
    agent_id: Optional[str] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
    state: Optional[Union[AgentState, dict]] = None,
    hooks: Optional[list[HookProvider]] = None,
    session_manager: Optional[SessionManager] = None,
    tool_executor: Optional[ToolExecutor] = None,
):
    """Initialize the Agent with the specified configuration.

    Args:
        model: Provider for running inference or a string representing the model-id for Bedrock to use.
            Defaults to strands.models.BedrockModel if None.
        messages: List of initial messages to pre-load into the conversation.
            Defaults to an empty list if None.
        tools: List of tools to make available to the agent.
            Can be specified as:

            - String tool names (e.g., "retrieve")
            - File paths (e.g., "/path/to/tool.py")
            - Imported Python modules (e.g., from strands_tools import current_time)
            - Dictionaries with name/path keys (e.g., {"name": "tool_name", "path": "/path/to/tool.py"})
            - ToolProvider instances for managed tool collections
            - Functions decorated with `@strands.tool` decorator.

            If provided, only these tools will be available. If None, all tools will be available.
        system_prompt: System prompt to guide model behavior.
            Can be a string or a list of SystemContentBlock objects for advanced features like caching.
            If None, the model will behave according to its default settings.
        structured_output_model: Pydantic model type(s) for structured output.
            When specified, all agent calls will attempt to return structured output of this type.
            This can be overridden on the agent invocation.
            Defaults to None (no structured output).
        callback_handler: Callback for processing events as they happen during agent execution.
            If not provided (using the default), a new PrintingCallbackHandler instance is created.
            If explicitly set to None, null_callback_handler is used.
        conversation_manager: Manager for conversation history and context window.
            Defaults to strands.agent.conversation_manager.SlidingWindowConversationManager if None.
        record_direct_tool_call: Whether to record direct tool calls in message history.
            Defaults to True.
        load_tools_from_directory: Whether to load and automatically reload tools in the `./tools/` directory.
            Defaults to False.
        trace_attributes: Custom trace attributes to apply to the agent's trace span.
        agent_id: Optional ID for the agent, useful for session management and multi-agent scenarios.
            Defaults to "default".
        name: name of the Agent
            Defaults to "Strands Agents".
        description: description of what the Agent does
            Defaults to None.
        state: stateful information for the agent. Can be either an AgentState object, or a json serializable dict.
            Defaults to an empty AgentState object.
        hooks: hooks to be added to the agent hook registry
            Defaults to None.
        session_manager: Manager for handling agent sessions including conversation history and state.
            If provided, enables session-based persistence and state management.
        tool_executor: Definition of tool execution strategy (e.g., sequential, concurrent, etc.).

    Raises:
        ValueError: If agent id contains path separators.
    """
    self.model = BedrockModel() if not model else BedrockModel(model_id=model) if isinstance(model, str) else model
    self.messages = messages if messages is not None else []
    # initializing self._system_prompt for backwards compatibility
    self._system_prompt, self._system_prompt_content = self._initialize_system_prompt(system_prompt)
    self._default_structured_output_model = structured_output_model
    self.agent_id = _identifier.validate(agent_id or _DEFAULT_AGENT_ID, _identifier.Identifier.AGENT)
    self.name = name or _DEFAULT_AGENT_NAME
    self.description = description

    # If not provided, create a new PrintingCallbackHandler instance
    # If explicitly set to None, use null_callback_handler
    # Otherwise use the passed callback_handler
    self.callback_handler: Union[Callable[..., Any], PrintingCallbackHandler]
    if isinstance(callback_handler, _DefaultCallbackHandlerSentinel):
        self.callback_handler = PrintingCallbackHandler()
    elif callback_handler is None:
        self.callback_handler = null_callback_handler
    else:
        self.callback_handler = callback_handler

    self.conversation_manager = conversation_manager if conversation_manager else SlidingWindowConversationManager()

    # Process trace attributes to ensure they're of compatible types
    self.trace_attributes: dict[str, AttributeValue] = {}
    if trace_attributes:
        for k, v in trace_attributes.items():
            if isinstance(v, (str, int, float, bool)) or (
                isinstance(v, list) and all(isinstance(x, (str, int, float, bool)) for x in v)
            ):
                self.trace_attributes[k] = v

    self.record_direct_tool_call = record_direct_tool_call
    self.load_tools_from_directory = load_tools_from_directory

    self.tool_registry = ToolRegistry()

    # Process tool list if provided
    if tools is not None:
        self.tool_registry.process_tools(tools)

    # Initialize tools and configuration
    self.tool_registry.initialize_tools(self.load_tools_from_directory)
    if load_tools_from_directory:
        self.tool_watcher = ToolWatcher(tool_registry=self.tool_registry)

    self.event_loop_metrics = EventLoopMetrics()

    # Initialize tracer instance (no-op if not configured)
    self.tracer = get_tracer()
    self.trace_span: Optional[trace_api.Span] = None

    # Initialize agent state management
    if state is not None:
        if isinstance(state, dict):
            self.state = AgentState(state)
        elif isinstance(state, AgentState):
            self.state = state
        else:
            raise ValueError("state must be an AgentState object or a dict")
    else:
        self.state = AgentState()

    self.tool_caller = _ToolCaller(self)

    self.hooks = HookRegistry()

    self._interrupt_state = _InterruptState()

    # Initialize session management functionality
    self._session_manager = session_manager
    if self._session_manager:
        self.hooks.add_hook(self._session_manager)

    # Allow conversation_managers to subscribe to hooks
    self.hooks.add_hook(self.conversation_manager)

    self.tool_executor = tool_executor or ConcurrentToolExecutor()

    if hooks:
        for hook in hooks:
            self.hooks.add_hook(hook)
    self.hooks.invoke_callbacks(AgentInitializedEvent(agent=self))

cleanup()

Clean up resources used by the agent.

This method cleans up all tool providers that require explicit cleanup, such as MCP clients. It should be called when the agent is no longer needed to ensure proper resource cleanup.

Note: This method uses a "belt and braces" approach with automatic cleanup through finalizers as a fallback, but explicit cleanup is recommended.

Source code in strands/agent/agent.py 
502
503
504
505
506
507
508
509
510
511
512
def cleanup(self) -> None:
    """Clean up resources used by the agent.

    This method cleans up all tool providers that require explicit cleanup,
    such as MCP clients. It should be called when the agent is no longer needed
    to ensure proper resource cleanup.

    Note: This method uses a "belt and braces" approach with automatic cleanup
    through finalizers as a fallback, but explicit cleanup is recommended.
    """
    self.tool_registry.cleanup()

invoke_async(prompt=None, *, invocation_state=None, structured_output_model=None, **kwargs) async

Process a natural language prompt through the agent's event loop.

This method implements the conversational interface with multiple input patterns: - String input: Simple text input - ContentBlock list: Multi-modal content blocks - Message list: Complete messages with roles - No input: Use existing conversation history

Parameters:

Name Type Description Default
prompt AgentInput

User input in various formats: - str: Simple text input - list[ContentBlock]: Multi-modal content blocks - list[Message]: Complete messages with roles - None: Use existing conversation history

 None
invocation_state dict[str, Any] | None

Additional parameters to pass through the event loop.

 None
structured_output_model Type[BaseModel] | None

Pydantic model type(s) for structured output (overrides agent default).

 None
**kwargs Any

Additional parameters to pass through the event loop.[Deprecating]

 {}

Returns:

Name Type Description
Result AgentResult

object containing:

  • stop_reason: Why the event loop stopped (e.g., "end_turn", "max_tokens")
  • message: The final message from the model
  • metrics: Performance metrics from the event loop
  • state: The final state of the event loop
Source code in strands/agent/agent.py 
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
async def invoke_async(
    self,
    prompt: AgentInput = None,
    *,
    invocation_state: dict[str, Any] | None = None,
    structured_output_model: Type[BaseModel] | None = None,
    **kwargs: Any,
) -> AgentResult:
    """Process a natural language prompt through the agent's event loop.

    This method implements the conversational interface with multiple input patterns:
    - String input: Simple text input
    - ContentBlock list: Multi-modal content blocks
    - Message list: Complete messages with roles
    - No input: Use existing conversation history

    Args:
        prompt: User input in various formats:
            - str: Simple text input
            - list[ContentBlock]: Multi-modal content blocks
            - list[Message]: Complete messages with roles
            - None: Use existing conversation history
        invocation_state: Additional parameters to pass through the event loop.
        structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
        **kwargs: Additional parameters to pass through the event loop.[Deprecating]

    Returns:
        Result: object containing:

            - stop_reason: Why the event loop stopped (e.g., "end_turn", "max_tokens")
            - message: The final message from the model
            - metrics: Performance metrics from the event loop
            - state: The final state of the event loop
    """
    events = self.stream_async(
        prompt, invocation_state=invocation_state, structured_output_model=structured_output_model, **kwargs
    )
    async for event in events:
        _ = event

    return cast(AgentResult, event["result"])

stream_async(prompt=None, *, invocation_state=None, structured_output_model=None, **kwargs) async

Process a natural language prompt and yield events as an async iterator.

This method provides an asynchronous interface for streaming agent events with multiple input patterns: - String input: Simple text input - ContentBlock list: Multi-modal content blocks - Message list: Complete messages with roles - No input: Use existing conversation history

Parameters:

Name Type Description Default
prompt AgentInput

User input in various formats: - str: Simple text input - list[ContentBlock]: Multi-modal content blocks - list[Message]: Complete messages with roles - None: Use existing conversation history

 None
invocation_state dict[str, Any] | None

Additional parameters to pass through the event loop.

 None
structured_output_model Type[BaseModel] | None

Pydantic model type(s) for structured output (overrides agent default).

 None
**kwargs Any

Additional parameters to pass to the event loop.[Deprecating]

 {}

Yields:

Type Description
AsyncIterator[Any]

An async iterator that yields events. Each event is a dictionary containing information about the current state of processing, such as:

  • data: Text content being generated
  • complete: Whether this is the final chunk
  • current_tool_use: Information about tools being executed
  • And other event data provided by the callback handler

Raises:

Type Description
Exception

Any exceptions from the agent invocation will be propagated to the caller.
Example 
async for event in agent.stream_async("Analyze this data"):
    if "data" in event:
        yield event["data"]
Source code in strands/agent/agent.py 
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
async def stream_async(
    self,
    prompt: AgentInput = None,
    *,
    invocation_state: dict[str, Any] | None = None,
    structured_output_model: Type[BaseModel] | None = None,
    **kwargs: Any,
) -> AsyncIterator[Any]:
    """Process a natural language prompt and yield events as an async iterator.

    This method provides an asynchronous interface for streaming agent events with multiple input patterns:
    - String input: Simple text input
    - ContentBlock list: Multi-modal content blocks
    - Message list: Complete messages with roles
    - No input: Use existing conversation history

    Args:
        prompt: User input in various formats:
            - str: Simple text input
            - list[ContentBlock]: Multi-modal content blocks
            - list[Message]: Complete messages with roles
            - None: Use existing conversation history
        invocation_state: Additional parameters to pass through the event loop.
        structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
        **kwargs: Additional parameters to pass to the event loop.[Deprecating]

    Yields:
        An async iterator that yields events. Each event is a dictionary containing
            information about the current state of processing, such as:

            - data: Text content being generated
            - complete: Whether this is the final chunk
            - current_tool_use: Information about tools being executed
            - And other event data provided by the callback handler

    Raises:
        Exception: Any exceptions from the agent invocation will be propagated to the caller.

    Example:
        ```python
        async for event in agent.stream_async("Analyze this data"):
            if "data" in event:
                yield event["data"]
        ```
    """
    self._interrupt_state.resume(prompt)

    self.event_loop_metrics.reset_usage_metrics()

    merged_state = {}
    if kwargs:
        warnings.warn("`**kwargs` parameter is deprecating, use `invocation_state` instead.", stacklevel=2)
        merged_state.update(kwargs)
        if invocation_state is not None:
            merged_state["invocation_state"] = invocation_state
    else:
        if invocation_state is not None:
            merged_state = invocation_state

    callback_handler = self.callback_handler
    if kwargs:
        callback_handler = kwargs.get("callback_handler", self.callback_handler)

    # Process input and get message to add (if any)
    messages = await self._convert_prompt_to_messages(prompt)

    self.trace_span = self._start_agent_trace_span(messages)

    with trace_api.use_span(self.trace_span):
        try:
            events = self._run_loop(messages, merged_state, structured_output_model)

            async for event in events:
                event.prepare(invocation_state=merged_state)

                if event.is_callback_event:
                    as_dict = event.as_dict()
                    callback_handler(**as_dict)
                    yield as_dict

            result = AgentResult(*event["stop"])
            callback_handler(result=result)
            yield AgentResultEvent(result=result).as_dict()

            self._end_agent_trace_span(response=result)

        except Exception as e:
            self._end_agent_trace_span(error=e)
            raise

structured_output(output_model, prompt=None)

This method allows you to get structured output from the agent.

If you pass in a prompt, it will be used temporarily without adding it to the conversation history. If you don't pass in a prompt, it will use only the existing conversation history to respond.

For smaller models, you may want to use the optional prompt to add additional instructions to explicitly instruct the model to output the structured data.

Parameters:

Name Type Description Default
output_model Type[T]

The output model (a JSON schema written as a Pydantic BaseModel) that the agent will use when responding.

 required
prompt AgentInput

The prompt to use for the agent in various formats: - str: Simple text input - list[ContentBlock]: Multi-modal content blocks - list[Message]: Complete messages with roles - None: Use existing conversation history

 None

Raises:

Type Description
ValueError

If no conversation history or prompt is provided.
Source code in strands/agent/agent.py 
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
def structured_output(self, output_model: Type[T], prompt: AgentInput = None) -> T:
    """This method allows you to get structured output from the agent.

    If you pass in a prompt, it will be used temporarily without adding it to the conversation history.
    If you don't pass in a prompt, it will use only the existing conversation history to respond.

    For smaller models, you may want to use the optional prompt to add additional instructions to explicitly
    instruct the model to output the structured data.

    Args:
        output_model: The output model (a JSON schema written as a Pydantic BaseModel)
            that the agent will use when responding.
        prompt: The prompt to use for the agent in various formats:
            - str: Simple text input
            - list[ContentBlock]: Multi-modal content blocks
            - list[Message]: Complete messages with roles
            - None: Use existing conversation history

    Raises:
        ValueError: If no conversation history or prompt is provided.
    """
    warnings.warn(
        "Agent.structured_output method is deprecated."
        " You should pass in `structured_output_model` directly into the agent invocation."
        " see: https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/structured-output/",
        category=DeprecationWarning,
        stacklevel=2,
    )

    return run_async(lambda: self.structured_output_async(output_model, prompt))

structured_output_async(output_model, prompt=None) async

This method allows you to get structured output from the agent.

If you pass in a prompt, it will be used temporarily without adding it to the conversation history. If you don't pass in a prompt, it will use only the existing conversation history to respond.

For smaller models, you may want to use the optional prompt to add additional instructions to explicitly instruct the model to output the structured data.

Parameters:

Name Type Description Default
output_model Type[T]

The output model (a JSON schema written as a Pydantic BaseModel) that the agent will use when responding.

 required
prompt AgentInput

The prompt to use for the agent (will not be added to conversation history).

 None

Raises:

Type Description
ValueError

If no conversation history or prompt is provided.

-

Source code in strands/agent/agent.py 
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
async def structured_output_async(self, output_model: Type[T], prompt: AgentInput = None) -> T:
    """This method allows you to get structured output from the agent.

    If you pass in a prompt, it will be used temporarily without adding it to the conversation history.
    If you don't pass in a prompt, it will use only the existing conversation history to respond.

    For smaller models, you may want to use the optional prompt to add additional instructions to explicitly
    instruct the model to output the structured data.

    Args:
        output_model: The output model (a JSON schema written as a Pydantic BaseModel)
            that the agent will use when responding.
        prompt: The prompt to use for the agent (will not be added to conversation history).

    Raises:
        ValueError: If no conversation history or prompt is provided.
    -
    """
    if self._interrupt_state.activated:
        raise RuntimeError("cannot call structured output during interrupt")

    warnings.warn(
        "Agent.structured_output_async method is deprecated."
        " You should pass in `structured_output_model` directly into the agent invocation."
        " see: https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/structured-output/",
        category=DeprecationWarning,
        stacklevel=2,
    )
    await self.hooks.invoke_callbacks_async(BeforeInvocationEvent(agent=self))
    with self.tracer.tracer.start_as_current_span(
        "execute_structured_output", kind=trace_api.SpanKind.CLIENT
    ) as structured_output_span:
        try:
            if not self.messages and not prompt:
                raise ValueError("No conversation history or prompt provided")

            temp_messages: Messages = self.messages + await self._convert_prompt_to_messages(prompt)

            structured_output_span.set_attributes(
                {
                    "gen_ai.system": "strands-agents",
                    "gen_ai.agent.name": self.name,
                    "gen_ai.agent.id": self.agent_id,
                    "gen_ai.operation.name": "execute_structured_output",
                }
            )
            if self.system_prompt:
                structured_output_span.add_event(
                    "gen_ai.system.message",
                    attributes={"role": "system", "content": serialize([{"text": self.system_prompt}])},
                )
            for message in temp_messages:
                structured_output_span.add_event(
                    f"gen_ai.{message['role']}.message",
                    attributes={"role": message["role"], "content": serialize(message["content"])},
                )
            events = self.model.structured_output(output_model, temp_messages, system_prompt=self.system_prompt)
            async for event in events:
                if isinstance(event, TypedEvent):
                    event.prepare(invocation_state={})
                    if event.is_callback_event:
                        self.callback_handler(**event.as_dict())

            structured_output_span.add_event(
                "gen_ai.choice", attributes={"message": serialize(event["output"].model_dump())}
            )
            return event["output"]

        finally:
            await self.hooks.invoke_callbacks_async(AfterInvocationEvent(agent=self))

StrandsA2AExecutor

Bases: AgentExecutor

Executor that adapts a Strands Agent to the A2A protocol.

This executor uses streaming mode to handle the execution of agent requests and converts Strands Agent responses to A2A protocol events.

Source code in strands/multiagent/a2a/executor.py 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
class StrandsA2AExecutor(AgentExecutor):
    """Executor that adapts a Strands Agent to the A2A protocol.

    This executor uses streaming mode to handle the execution of agent requests
    and converts Strands Agent responses to A2A protocol events.
    """

    # Default formats for each file type when MIME type is unavailable or unrecognized
    DEFAULT_FORMATS = {"document": "txt", "image": "png", "video": "mp4", "unknown": "txt"}

    # Handle special cases where format differs from extension
    FORMAT_MAPPINGS = {"jpg": "jpeg", "htm": "html", "3gp": "three_gp", "3gpp": "three_gp", "3g2": "three_gp"}

    def __init__(self, agent: SAAgent):
        """Initialize a StrandsA2AExecutor.

        Args:
            agent: The Strands Agent instance to adapt to the A2A protocol.
        """
        self.agent = agent

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        """Execute a request using the Strands Agent and send the response as A2A events.

        This method executes the user's input using the Strands Agent in streaming mode
        and converts the agent's response to A2A events.

        Args:
            context: The A2A request context, containing the user's input and task metadata.
            event_queue: The A2A event queue used to send response events back to the client.

        Raises:
            ServerError: If an error occurs during agent execution
        """
        task = context.current_task
        if not task:
            task = new_task(context.message)  # type: ignore
            await event_queue.enqueue_event(task)

        updater = TaskUpdater(event_queue, task.id, task.context_id)

        try:
            await self._execute_streaming(context, updater)
        except Exception as e:
            raise ServerError(error=InternalError()) from e

    async def _execute_streaming(self, context: RequestContext, updater: TaskUpdater) -> None:
        """Execute request in streaming mode.

        Streams the agent's response in real-time, sending incremental updates
        as they become available from the agent.

        Args:
            context: The A2A request context, containing the user's input and other metadata.
            updater: The task updater for managing task state and sending updates.
        """
        # Convert A2A message parts to Strands ContentBlocks
        if context.message and hasattr(context.message, "parts"):
            content_blocks = self._convert_a2a_parts_to_content_blocks(context.message.parts)
            if not content_blocks:
                raise ValueError("No content blocks available")
        else:
            raise ValueError("No content blocks available")

        try:
            async for event in self.agent.stream_async(content_blocks):
                await self._handle_streaming_event(event, updater)
        except Exception:
            logger.exception("Error in streaming execution")
            raise

    async def _handle_streaming_event(self, event: dict[str, Any], updater: TaskUpdater) -> None:
        """Handle a single streaming event from the Strands Agent.

        Processes streaming events from the agent, converting data chunks to A2A
        task updates and handling the final result when streaming is complete.

        Args:
            event: The streaming event from the agent, containing either 'data' for
                incremental content or 'result' for the final response.
            updater: The task updater for managing task state and sending updates.
        """
        logger.debug("Streaming event: %s", event)
        if "data" in event:
            if text_content := event["data"]:
                await updater.update_status(
                    TaskState.working,
                    new_agent_text_message(
                        text_content,
                        updater.context_id,
                        updater.task_id,
                    ),
                )
        elif "result" in event:
            await self._handle_agent_result(event["result"], updater)

    async def _handle_agent_result(self, result: SAAgentResult | None, updater: TaskUpdater) -> None:
        """Handle the final result from the Strands Agent.

        Processes the agent's final result, extracts text content from the response,
        and adds it as an artifact to the task before marking the task as complete.

        Args:
            result: The agent result object containing the final response, or None if no result.
            updater: The task updater for managing task state and adding the final artifact.
        """
        if final_content := str(result):
            await updater.add_artifact(
                [Part(root=TextPart(text=final_content))],
                name="agent_response",
            )
        await updater.complete()

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        """Cancel an ongoing execution.

        This method is called when a request cancellation is requested. Currently,
        cancellation is not supported by the Strands Agent executor, so this method
        always raises an UnsupportedOperationError.

        Args:
            context: The A2A request context.
            event_queue: The A2A event queue.

        Raises:
            ServerError: Always raised with an UnsupportedOperationError, as cancellation
                is not currently supported.
        """
        logger.warning("Cancellation requested but not supported")
        raise ServerError(error=UnsupportedOperationError())

    def _get_file_type_from_mime_type(self, mime_type: str | None) -> Literal["document", "image", "video", "unknown"]:
        """Classify file type based on MIME type.

        Args:
            mime_type: The MIME type of the file

        Returns:
            The classified file type
        """
        if not mime_type:
            return "unknown"

        mime_type = mime_type.lower()

        if mime_type.startswith("image/"):
            return "image"
        elif mime_type.startswith("video/"):
            return "video"
        elif (
            mime_type.startswith("text/")
            or mime_type.startswith("application/")
            or mime_type in ["application/pdf", "application/json", "application/xml"]
        ):
            return "document"
        else:
            return "unknown"

    def _get_file_format_from_mime_type(self, mime_type: str | None, file_type: str) -> str:
        """Extract file format from MIME type using Python's mimetypes library.

        Args:
            mime_type: The MIME type of the file
            file_type: The classified file type (image, video, document, txt)

        Returns:
            The file format string
        """
        if not mime_type:
            return self.DEFAULT_FORMATS.get(file_type, "txt")

        mime_type = mime_type.lower()

        # Extract subtype from MIME type and check existing format mappings
        if "/" in mime_type:
            subtype = mime_type.split("/")[-1]
            if subtype in self.FORMAT_MAPPINGS:
                return self.FORMAT_MAPPINGS[subtype]

        # Use mimetypes library to find extensions for the MIME type
        extensions = mimetypes.guess_all_extensions(mime_type)

        if extensions:
            extension = extensions[0][1:]  # Remove the leading dot
            return self.FORMAT_MAPPINGS.get(extension, extension)

        # Fallback to defaults for unknown MIME types
        return self.DEFAULT_FORMATS.get(file_type, "txt")

    def _strip_file_extension(self, file_name: str) -> str:
        """Strip the file extension from a file name.

        Args:
            file_name: The original file name with extension

        Returns:
            The file name without extension
        """
        if "." in file_name:
            return file_name.rsplit(".", 1)[0]
        return file_name

    def _convert_a2a_parts_to_content_blocks(self, parts: list[Part]) -> list[ContentBlock]:
        """Convert A2A message parts to Strands ContentBlocks.

        Args:
            parts: List of A2A Part objects

        Returns:
            List of Strands ContentBlock objects
        """
        content_blocks: list[ContentBlock] = []

        for part in parts:
            try:
                part_root = part.root

                if isinstance(part_root, TextPart):
                    # Handle TextPart
                    content_blocks.append(ContentBlock(text=part_root.text))

                elif isinstance(part_root, FilePart):
                    # Handle FilePart
                    file_obj = part_root.file
                    mime_type = getattr(file_obj, "mime_type", None)
                    raw_file_name = getattr(file_obj, "name", "FileNameNotProvided")
                    file_name = self._strip_file_extension(raw_file_name)
                    file_type = self._get_file_type_from_mime_type(mime_type)
                    file_format = self._get_file_format_from_mime_type(mime_type, file_type)

                    # Handle FileWithBytes vs FileWithUri
                    bytes_data = getattr(file_obj, "bytes", None)
                    uri_data = getattr(file_obj, "uri", None)

                    if bytes_data:
                        try:
                            # A2A bytes are always base64-encoded strings
                            decoded_bytes = base64.b64decode(bytes_data)
                        except Exception as e:
                            raise ValueError(f"Failed to decode base64 data for file '{raw_file_name}': {e}") from e

                        if file_type == "image":
                            content_blocks.append(
                                ContentBlock(
                                    image=ImageContent(
                                        format=file_format,  # type: ignore
                                        source=ImageSource(bytes=decoded_bytes),
                                    )
                                )
                            )
                        elif file_type == "video":
                            content_blocks.append(
                                ContentBlock(
                                    video=VideoContent(
                                        format=file_format,  # type: ignore
                                        source=VideoSource(bytes=decoded_bytes),
                                    )
                                )
                            )
                        else:  # document or unknown
                            content_blocks.append(
                                ContentBlock(
                                    document=DocumentContent(
                                        format=file_format,  # type: ignore
                                        name=file_name,
                                        source=DocumentSource(bytes=decoded_bytes),
                                    )
                                )
                            )
                    # Handle FileWithUri
                    elif uri_data:
                        # For URI files, create a text representation since Strands ContentBlocks expect bytes
                        content_blocks.append(
                            ContentBlock(
                                text="[File: %s (%s)] - Referenced file at: %s" % (file_name, mime_type, uri_data)
                            )
                        )
                elif isinstance(part_root, DataPart):
                    # Handle DataPart - convert structured data to JSON text
                    try:
                        data_text = json.dumps(part_root.data, indent=2)
                        content_blocks.append(ContentBlock(text="[Structured Data]\n%s" % data_text))
                    except Exception:
                        logger.exception("Failed to serialize data part")
            except Exception:
                logger.exception("Error processing part")

        return content_blocks

__init__(agent)

Initialize a StrandsA2AExecutor.

Parameters:

Name Type Description Default
agent Agent

The Strands Agent instance to adapt to the A2A protocol.

 required
Source code in strands/multiagent/a2a/executor.py 
52
53
54
55
56
57
58
def __init__(self, agent: SAAgent):
    """Initialize a StrandsA2AExecutor.

    Args:
        agent: The Strands Agent instance to adapt to the A2A protocol.
    """
    self.agent = agent

cancel(context, event_queue) async

Cancel an ongoing execution.

This method is called when a request cancellation is requested. Currently, cancellation is not supported by the Strands Agent executor, so this method always raises an UnsupportedOperationError.

Parameters:

Name Type Description Default
context RequestContext

The A2A request context.

 required
event_queue EventQueue

The A2A event queue.

 required

Raises:

Type Description
ServerError

Always raised with an UnsupportedOperationError, as cancellation is not currently supported.
Source code in strands/multiagent/a2a/executor.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
    """Cancel an ongoing execution.

    This method is called when a request cancellation is requested. Currently,
    cancellation is not supported by the Strands Agent executor, so this method
    always raises an UnsupportedOperationError.

    Args:
        context: The A2A request context.
        event_queue: The A2A event queue.

    Raises:
        ServerError: Always raised with an UnsupportedOperationError, as cancellation
            is not currently supported.
    """
    logger.warning("Cancellation requested but not supported")
    raise ServerError(error=UnsupportedOperationError())

execute(context, event_queue) async

Execute a request using the Strands Agent and send the response as A2A events.

This method executes the user's input using the Strands Agent in streaming mode and converts the agent's response to A2A events.

Parameters:

Name Type Description Default
context RequestContext

The A2A request context, containing the user's input and task metadata.

 required
event_queue EventQueue

The A2A event queue used to send response events back to the client.

 required

Raises:

Type Description
ServerError

If an error occurs during agent execution
Source code in strands/multiagent/a2a/executor.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
async def execute(
    self,
    context: RequestContext,
    event_queue: EventQueue,
) -> None:
    """Execute a request using the Strands Agent and send the response as A2A events.

    This method executes the user's input using the Strands Agent in streaming mode
    and converts the agent's response to A2A events.

    Args:
        context: The A2A request context, containing the user's input and task metadata.
        event_queue: The A2A event queue used to send response events back to the client.

    Raises:
        ServerError: If an error occurs during agent execution
    """
    task = context.current_task
    if not task:
        task = new_task(context.message)  # type: ignore
        await event_queue.enqueue_event(task)

    updater = TaskUpdater(event_queue, task.id, task.context_id)

    try:
        await self._execute_streaming(context, updater)
    except Exception as e:
        raise ServerError(error=InternalError()) from e