# A2aAgentServer ## `autogen.a2a.A2aAgentServer` ``` A2aAgentServer(agent, *, url='http://localhost:8000', agent_card=None, card_modifier=None, extended_agent_card=None, extended_card_modifier=None) ``` A server wrapper for running an AG2 agent via the A2A protocol. This class provides functionality to wrap an AG2 ConversableAgent into an A2A server that can be used to interact with the agent through A2A requests. Initialize the A2aAgentServer. | PARAMETER | DESCRIPTION | |------------------------|-------------| | `agent` | The Autogen ConversableAgent to serve.
**TYPE:**`ConversableAgent` | | `url` | The base URL for the A2A server.
**TYPE:**`str | None`**DEFAULT:**`'http://localhost:8000'` | | `agent_card` | Configuration for the base agent card.
**TYPE:**`CardSettings | None`**DEFAULT:**`None` | | `card_modifier` | Function to modify the base agent card.
**TYPE:**`Callable[[AgentCard], AgentCard] | None`**DEFAULT:**`None` | | `extended_agent_card` | Configuration for the extended agent card.
**TYPE:**`CardSettings | None`**DEFAULT:**`None` | | `extended_card_modifier`| Function to modify the extended agent card.
**TYPE:**`Callable[[AgentCard, ServerCallContext], AgentCard] | None`**DEFAULT:**`None` | Source code in `autogen/a2a/server.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
``` | ```
def __init__(
self,
agent: "ConversableAgent",
*,
url: str | None = "http://localhost:8000",
agent_card: CardSettings | None = None,
card_modifier: Callable[["AgentCard"], "AgentCard"] | None = None,
extended_agent_card: CardSettings | None = None,
extended_card_modifier: Callable[["AgentCard", "ServerCallContext"], "AgentCard"] | None = None,
) -> None:
"""Initialize the A2aAgentServer.
Args:
agent: The Autogen ConversableAgent to serve.
url: The base URL for the A2A server.
agent_card: Configuration for the base agent card.
card_modifier: Function to modify the base agent card.
extended_agent_card: Configuration for the extended agent card.
extended_card_modifier: Function to modify the extended agent card.
"""
self.agent = agent
if not agent_card:
agent_card = CardSettings()
if agent_card.url and url != "http://localhost:8000":
warnings.warn(
(
"You can't use `agent_card.url` and `url` options in the same time. "
f"`agent_card.url` has a higher priority, so {{agent_card.url}} will be used."
),
RuntimeWarning,
stacklevel=2,
)
self.card = AgentCard.model_validate({
# use agent options by default
"name": agent.name,
"description": agent.description,
"url": url,
"supports_authenticated_extended_card": extended_agent_card is not None,
# exclude name and description if not provided
**agent_card.model_dump(exclude_none=True),
})
self.extended_agent_card: AgentCard | None = None
if extended_agent_card:
if extended_agent_card.url and url != "http://localhost:8000":
warnings.warn(
(
"You can't use `extended_agent_card.url` and `url` options in the same time. "
f"`agent_card.url` has a higher priority, so {{extended_agent_card.url}} will be used."
),
RuntimeWarning,
stacklevel=2,
)
self.extended_agent_card = AgentCard.model_validate({
"name": agent.name,
"description": agent.description,
"url": url,
**extended_agent_card.model_dump(exclude_none=True),
})
self.card_modifier = card_modifier
self.extended_card_modifier = extended_card_modifier
``` | ### `agent` instance-attribute ``` agents = agent ``` ### `card` instance-attribute ``` card = model_validate({'name': name, 'description': description, 'url': url, 'supports_authenticated_extended_card': extended_agent_card is not None, None: model_dump(exclude_none=True)}) ``` ### `extended_agent_card` instance-attribute ``` extended_agent_card = None ``` ### `card_modifier` instance-attribute ``` card_modifier = card_modifier ``` ### `extended_card_modifier` instance-attribute ``` extended_card_modifier = extended_card_modifier ``` ### `executor` property ``` executor ``` Get the A2A agent executor. ### `build` class-attribute instance-attribute ``` build = build_starlette_app ``` ### `build_request_handler` ``` build_request_handler(*, task_store=None, queue_manager=None, push_config_store=None, push_sender=None, request_context_builder=None) ``` Build a request handler for A2A application. | PARAMETER | DESCRIPTION | |------------------------|-------------| | `task_store` | The task store to use.
**TYPE:**`TaskStore | None`**DEFAULT:**`None` | | `queue_manager` | The queue manager to use.
**TYPE:**`QueueManager | None`**DEFAULT:**`None` | | `push_config_store` | The push notification config store to use.
**TYPE:**`PushNotificationConfigStore | None`**DEFAULT:**`None` | | `push_sender` | The push notification sender to use.
**TYPE:**`PushNotificationSender | None`**DEFAULT:**`None` | | `request_context_builder` | The request context builder to use.
**TYPE:**`RequestContextBuilder | None`**DEFAULT:**`None` | | RETURNS | DESCRIPTION | |-----------------------|-------------| | `RequestHandler` | A configured RequestHandler instance. | Source code in `autogen/a2a/server.py` | | | | --- | --- | | ```
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
``` | ```
def build_request_handler(
self,
*,
task_store: "TaskStore | None" = None,
queue_manager: "QueueManager | None" = None,
push_config_store: "PushNotificationConfigStore | None" = None,
push_sender: "PushNotificationSender | None" = None,
request_context_builder: "RequestContextBuilder | None" = None,
) -> "RequestHandler":
"""Build a request handler for A2A application.
Args:
task_store: The task store to use.
queue_manager: The queue manager to use.
push_config_store: The push notification config store to use.
push_sender: The push notification sender to use.
request_context_builder: The request context builder to use.
Returns:
A configured RequestHandler instance.
"""
return DefaultRequestHandler(
agent_executor=self.executor,
task_store=task_store or InMemoryTaskStore(),
queue_manager=queue_manager,
push_config_store=push_config_store,
push_sender=push_sender,
request_context_builder=request_context_builder,
)
``` | ### `build_starlette_app` ``` build_starlette_app(*, request_handler=None, context_builder=None) ``` Build a Starlette A2A application for ASGI server. | PARAMETER | DESCRIPTION | |------------------------|-------------| | `request_handler` | The request handler to use.
**TYPE:**`RequestHandler | None`**DEFAULT:**`None` | | `context_builder` | The context builder to use.
**TYPE:**`CallContextBuilder | None`**DEFAULT:**`None` | | RETURNS | DESCRIPTION | |------------------------|-------------| | `Starlette` | A configured Starlette application instance. | Source code in `autogen/a2a/server.py` | | | | --- | --- | | ```
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
``` | ```
def build_starlette_app(
self,
*,
request_handler: "RequestHandler | None" = None,
context_builder: "CallContextBuilder | None" = None,
) -> "Starlette":
"""Build a Starlette A2A application for ASGI server.
Args:
request_handler: The request handler to use.
context_builder: The context builder to use.
Returns:
A configured Starlette application instance.
"""
from a2a.server.apps import A2AStarletteApplication
return A2AStarletteApplication(
agent_card=self.card,
extended_agent_card=self.extended_agent_card,
http_handler=request_handler
or DefaultRequestHandler(
agent_executor=self.executor,
task_store=InMemoryTaskStore(),
),
context_builder=context_builder,
card_modifier=self.card_modifier,
extended_card_modifier=self.extended_card_modifier,
).build()
``` |