Skip to content

ant_ai.a2a.server

A2AServer pydantic-model

Bases: BaseModel

Class to instantiate a fully usable A2A server via uvicorn, either as a FastAPI or Starlete app.

Config:

  • arbitrary_types_allowed: True

Fields:

  • agent (Agent)
  • workflow (Workflow)
  • host (str)
  • port (int)
  • agent_card (AgentCard)
  • task_store (TaskStore)
  • queue_manager (QueueManager | None)
  • push_config_store (PushNotificationConfigStore | None)
  • push_sender (PushNotificationSender | None)
  • stream_artifacts (bool)
  • context_class (type[InvocationContext])

Validators:

  • _create_request_handler
Source code in src/ant_ai/a2a/server.py 
 25
 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
class A2AServer(BaseModel):
    """Class to instantiate a fully usable A2A server via uvicorn, either as a FastAPI or Starlete app."""

    agent: Agent
    workflow: Workflow
    host: str = Field(default="127.0.0.1")
    port: int = Field(default=9000)
    agent_card: AgentCard
    task_store: TaskStore = Field(default_factory=InMemoryTaskStore)
    queue_manager: QueueManager | None = Field(default=None)
    push_config_store: PushNotificationConfigStore | None = Field(default=None)
    push_sender: PushNotificationSender | None = Field(default=None)
    stream_artifacts: bool = Field(
        default=True,
        description=(
            "Translate ContentDeltaEvent into A2A TaskArtifactUpdateEvent "
            "chunks via add_artifact, forwarding live token deltas to "
            "whatever peer is talking to this server. The agent's own "
            "stream() always emits ContentDeltaEvents internally regardless "
            "of this flag -- this only decides whether they also go out "
            "over the wire."
        ),
    )

    context_class: type[InvocationContext] = Field(
        default=InvocationContext,
        description=(
            "The InvocationContext (sub)class built for each request from the A2A message metadata. Subclass InvocationContext to carry your own fields through the run; they are filled from metadata by name."
        ),
    )

    model_config = ConfigDict(arbitrary_types_allowed=True)

    @model_validator(mode="after")
    def _create_request_handler(self) -> A2AServer:
        request_context_builder = HistoryRequestContextBuilder(
            should_populate_referred_tasks=True,
            task_store=self.task_store,
        )

        self._request_handler: DefaultRequestHandler = DefaultRequestHandler(
            agent_executor=A2AExecutor(
                agent=self.agent,
                workflow=self.workflow,
                stream_artifacts=self.stream_artifacts,
                context_class=self.context_class,
            ),
            task_store=self.task_store,
            agent_card=self.agent_card,
            request_context_builder=request_context_builder,
            queue_manager=self.queue_manager,
            push_config_store=self.push_config_store,
            push_sender=self.push_sender,
        )
        return self

    def _build_routes(self) -> list:
        routes = []
        routes.extend(create_agent_card_routes(self.agent_card))
        routes.extend(create_jsonrpc_routes(self._request_handler, rpc_url="/"))
        return routes

    def starlette_app(self) -> Starlette:
        """Create a Starlette application for serving the agent"""
        return Starlette(routes=self._build_routes())

    def fastapi_app(self) -> FastAPI:
        """Create a FastAPI application for serving the agent"""
        return FastAPI(title=self.agent_card.name, routes=self._build_routes())

    def serve(self, use_fastapi: bool = True) -> None:
        """Start serving the agent using Uvicorn

        Args:
            use_fastapi: Whether to use FastAPI or Starlette as the web framework. Defaults to True (FastAPI).

        Returns:
            None
        """
        try:
            import uvicorn

            logger.info(
                f"Starting A2A server for agent '{self.agent.name}' at {self.host}:{self.port}..."
            )
            app: Starlette = self.fastapi_app() if use_fastapi else self.starlette_app()
            logger.info(
                f"Using {'FastAPI' if use_fastapi else 'Starlette'} application"
            )
            uvicorn.run(app, host=self.host, port=self.port)
        except ImportError as e:
            raise ImportError(
                "Uvicorn is not installed. Please install it with 'pip install uvicorn'."
            ) from e
        except KeyboardInterrupt:
            logger.info("Server stopped")
        except Exception as e:
            logger.error(f"Failed to start server: {e}")
            raise RuntimeError(f"Failed to start the server: {e}") from e

stream_artifacts pydantic-field 

stream_artifacts: bool = True

Translate ContentDeltaEvent into A2A TaskArtifactUpdateEvent chunks via add_artifact, forwarding live token deltas to whatever peer is talking to this server. The agent's own stream() always emits ContentDeltaEvents internally regardless of this flag -- this only decides whether they also go out over the wire.

context_class pydantic-field 

context_class: type[InvocationContext] = InvocationContext

The InvocationContext (sub)class built for each request from the A2A message metadata. Subclass InvocationContext to carry your own fields through the run; they are filled from metadata by name.

starlette_app 

starlette_app() -> Starlette

Create a Starlette application for serving the agent

Source code in src/ant_ai/a2a/server.py 
87
88
89
def starlette_app(self) -> Starlette:
    """Create a Starlette application for serving the agent"""
    return Starlette(routes=self._build_routes())

fastapi_app 

fastapi_app() -> FastAPI

Create a FastAPI application for serving the agent

Source code in src/ant_ai/a2a/server.py 
91
92
93
def fastapi_app(self) -> FastAPI:
    """Create a FastAPI application for serving the agent"""
    return FastAPI(title=self.agent_card.name, routes=self._build_routes())

serve 

serve(use_fastapi: bool = True) -> None

Start serving the agent using Uvicorn

Parameters:

Name Type Description Default
use_fastapi bool

Whether to use FastAPI or Starlette as the web framework. Defaults to True (FastAPI).

 True

Returns:

Type Description
None

None
Source code in src/ant_ai/a2a/server.py 
 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
def serve(self, use_fastapi: bool = True) -> None:
    """Start serving the agent using Uvicorn

    Args:
        use_fastapi: Whether to use FastAPI or Starlette as the web framework. Defaults to True (FastAPI).

    Returns:
        None
    """
    try:
        import uvicorn

        logger.info(
            f"Starting A2A server for agent '{self.agent.name}' at {self.host}:{self.port}..."
        )
        app: Starlette = self.fastapi_app() if use_fastapi else self.starlette_app()
        logger.info(
            f"Using {'FastAPI' if use_fastapi else 'Starlette'} application"
        )
        uvicorn.run(app, host=self.host, port=self.port)
    except ImportError as e:
        raise ImportError(
            "Uvicorn is not installed. Please install it with 'pip install uvicorn'."
        ) from e
    except KeyboardInterrupt:
        logger.info("Server stopped")
    except Exception as e:
        logger.error(f"Failed to start server: {e}")
        raise RuntimeError(f"Failed to start the server: {e}") from e