# 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.<br>**TYPE:**`ConversableAgent`                                   |
| `url`                     | The base URL for the A2A server.<br>**TYPE:**`str | None`**DEFAULT:**`'http://localhost:8000'`          |
| `agent_card`              | Configuration for the base agent card.<br>**TYPE:**`CardSettings | None`**DEFAULT:**`None`             |
| `card_modifier`           | Function to modify the base agent card.<br>**TYPE:**`Callable[[AgentCard], AgentCard] | None`**DEFAULT:**`None`|
| `extended_agent_card`     | Configuration for the extended agent card.<br>**TYPE:**`CardSettings | None`**DEFAULT:**`None`        |
| `extended_card_modifier`   | Function to modify the extended agent card.<br>**TYPE:**`Callable[[AgentCard, ServerCallContext], AgentCard] | None`**DEFAULT:**`None`|

Source code in `autogen/a2a/server.py`

```python

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({
        "name": agent.name,
        "description": agent.description,
        "url": url,
        "supports_authenticated_extended_card": extended_agent_card is not None,
        **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),
        })
    try:
        from autogen.agents.experimental.a2ui import A2UIAgent
        from autogen.agents.experimental.a2ui.a2a_helpers import A2UI_EXTENSION_URI
        if isinstance(agent, A2UIAgent):
            existing_extensions = list(self.card.capabilities.extensions or [])
            if not any(e.uri == A2UI_EXTENSION_URI for e in existing_extensions):
                params: dict[str, Any] = {
                    "supportedCatalogIds": [agent.catalog_id],
                }
                existing_extensions.append(
                    AgentExtension(
                        uri=A2UI_EXTENSION_URI,
                        description="Provides agent-driven UI using the A2UI v0.9 JSON format.",
                        params=params,
                    )
                )
                self.card.capabilities.extensions = existing_extensions
    except (ImportError, NameError):
        pass
    self.card_modifier = card_modifier
    self.extended_card_modifier = extended_card_modifier
    self.middlewares: list[tuple[BaseHTTPMiddleware, dict[str, Any]]] = []
```

### `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
```

### `middlewares` instance-attribute

```
middlewares = []
```

### `executor` property

Get the A2A agent executor.

Auto-detects `A2UIAgent` and returns an `A2UIAgentExecutor` that preserves A2UI DataParts in responses and handles extension negotiation.

### `build` class-attribute instance-attribute

```
build = build_starlette_app
```

### `add_middleware`

```
add_middleware(middleware, **kwargs)
```

Add a middleware to the A2A server.

Source code in `autogen/a2a/server.py`

```python
def add_middleware(self, middleware: "BaseHTTPMiddleware", **kwargs: Any) -> None:
    """Add a middleware to the A2A server."""
    self.middlewares.append((middleware, kwargs))
```

### `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.<br>**TYPE:**`TaskStore | None`**DEFAULT:**`None`                                   |
| `queue_manager`            | The queue manager to use.<br>**TYPE:**`QueueManager | None`**DEFAULT:**`None`                                |
| `push_config_store`        | The push notification config store to use.<br>**TYPE:**`PushNotificationConfigStore | None`**DEFAULT:**`None`|
| `push_sender`              | The push notification sender to use.<br>**TYPE:**`PushNotificationSender | None`**DEFAULT:**`None`         |
| `request_context_builder`   | The request context builder to use.<br>**TYPE:**`RequestContextBuilder | None`**DEFAULT:**`None`           |

| RETURNS                     | DESCRIPTION                                                                                                 |
|----------------------------|-------------------------------------------------------------------------------------------------------------|
| `RequestHandler`           | A configured RequestHandler instance.                                                                       |

Source code in `autogen/a2a/server.py`

```python
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.<br>**TYPE:**`RequestHandler | None`**DEFAULT:**`None`                           |
| `context_builder`          | The context builder to use.<br>**TYPE:**`CallContextBuilder | None`**DEFAULT:**`None`                       |

| RETURNS                     | DESCRIPTION                                                                                                 |
|----------------------------|-------------------------------------------------------------------------------------------------------------|
| `Starlette`                | A configured Starlette application instance.                                                               |

Source code in `autogen/a2a/server.py`

```python
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
    app = 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()
    for middleware, kwargs in self.middlewares:
        app.add_middleware(middleware, **kwargs)  # type: ignore[arg-type]
    return app
```