Function create_server_app

tests/unittests/a2a/integration/server.py:76–101

Creates an A2A FastAPI application with a mocked runner. Args: run_async_fn: A generator function that takes **kwargs and yields Event objects. config: Optional executor configuration. task_store: Optional task store instance. Defaults to InMemoryTaskStore. Returns: A Fas

(
    run_async_fn=None,
    config: A2aAgentExecutorConfig | None = None,
    task_store=None,
)

def create_server_app(
    run_async_fn=None,
    config: A2aAgentExecutorConfig | None = None,
    task_store=None,
):
  """Creates an A2A FastAPI application with a mocked runner.

  Args:
    run_async_fn: A generator function that takes **kwargs and yields Event
      objects.
    config: Optional executor configuration.
    task_store: Optional task store instance. Defaults to InMemoryTaskStore.

  Returns:
    A FastAPI application instance.
  """
  runner = FakeRunner(run_async_fn)
  executor = A2aAgentExecutor(runner=runner, config=config)
  if task_store is None:
    task_store = InMemoryTaskStore()
  handler = DefaultRequestHandler(
      agent_executor=executor, task_store=task_store
  )

  app = A2AFastAPIApplication(agent_card=agent_card, http_handler=handler)
  return app.build()