Skip to content

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 #

agent = 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()