Remarque
L’accès à cette page nécessite une autorisation. Vous pouvez essayer de vous connecter ou de modifier des répertoires.
L’accès à cette page nécessite une autorisation. Vous pouvez essayer de modifier des répertoires.
Utilisez les packages d’hébergement .NET A2A pour exposer un agent Agent Framework via ASP.NET Core. Consultez les agents hôtes avec A2A pour obtenir un guide complet sur le serveur multi language.
Utilisez le package Go provider/a2aprovider avec les gestionnaires de serveurs A2A Go officiels. Consultez les agents hôtes avec A2A pour obtenir un exemple complet de serveur.
Agent Framework fournit deux packages Python pour l’hébergement d’agents et de flux de travail via le SDK A2A officiel :
| Package | Modèle d’intégration | Utilisez-le quand |
|---|---|---|
agent-framework-a2a |
Un A2AExecutor préconfiguré qui convertit les requêtes, exécute un agent et publie des événements et des artefacts de tâche A2A. |
Vous souhaitez le comportement standard entre Agent Framework et A2A et avez seulement besoin d’assembler le serveur SDK A2A. |
agent-framework-hosting-a2a |
Composants de base incrémentiels pour un moteur d’exécution propre à l’application. Commencez par l’agent de base ou les convertisseurs de flux de travail, et utilisez AgentA2AAdapter éventuellement ou WorkflowA2AAdapter, si vous le souhaitez, qui s’appuient sur ces convertisseurs pour ajouter la génération de cartes natives et la validation du mode. |
Votre application doit posséder un mappage de session, des transitions de tâches, une remise d’événements, des limites d’artefact, une conversion de sortie ou un hôte multi-protocole. |
Les deux packages utilisent des types de SDK A2A natifs et des composants serveur. Votre application fournit le gestionnaire de demandes, le magasin de tâches, les itinéraires ou le générateur d’applications sdk, l’authentification et le déploiement. Avec agent-framework-hosting-a2a, l’application peut construire la carte d’agent directement ou laisser un adaptateur le générer.
Utiliser l’exécuteur A2A préconfiguré
Installez agent-framework-a2a lorsque l’adaptateur serveur intégré correspond à votre cycle de vie :
pip install --pre agent-framework-a2a starlette uvicorn
A2AExecutor implémente le AgentExecutor du SDK A2A. Il lit l’entrée de l’utilisateur à partir du contexte de requête A2A, crée une session Agent Framework à partir de l’ID de contexte A2A, exécute l’agent en mode streaming ou non-streaming, convertit le contenu de sortie pris en charge et publie l’état des tâches et les événements d’artefact TaskUpdatervia le SDK.
Composez-le avec l’application A2A SDK, le magasin de tâches, la carte de l’agent DefaultRequestHandleret l’application Starlette ou une autre intégration de serveur prise en charge. Configurez la diffusion en continu avec A2AExecutor(agent, stream=True), transmettez des options stables d’exécution de l’agent via run_kwargs, ou sous-classez A2AExecutor et redéfinissez handle_events lorsque vous avez besoin d’un mappage de sortie différent.
A2AExecutor est limité à un point de terminaison A2A et gère directement son mappage d’exécution et de session A2A. Utilisez les packages d’hébergement lorsque le même agent doit être disponible via plusieurs protocoles dans une application.
Pour obtenir la configuration complète du serveur, consultez Exposer un agent Agent Framework sur A2A.
Utiliser un adaptateur dans un exécuteur appartenant à l’application
Installez le package d’hébergement lorsque votre application possède l’exécuteur A2A natif, mais souhaite que Agent Framework génère la carte publique et valide les conversions :
pip install --pre agent-framework-hosting-a2a starlette uvicorn
AgentA2AAdapter accepte un agent ou AgentState. Sa méthode asynchrone get_card dérive le nom et la description publics, utilise les modes de texte conservateurs par défaut et peut déduire des compétences A2A natives à partir d’instances Agent Framework SkillsProvider . Les fonctionnalités du serveur et les interfaces prises en charge restent explicites, car elles décrivent le point de terminaison de l’application plutôt que la méthode de l’agent run .
L’adaptateur expose les méthodes a2a_to_run et a2a_from_run, qui valident par défaut les valeurs par rapport aux modes de carte configurés. L’application possède toujours l’exécuteur A2A, le cycle de vie des tâches, la file d’attente d’événements, les limites des artefacts, la stratégie de session, l’authentification, les itinéraires et le déploiement.
Cet exécuteur utilise un adaptateur pour la conversion entrante, l’état de l’agent et la conversion sortante :
class AppAgentExecutor(AgentExecutor):
"""Native A2A SDK executor composed with Agent Framework conversion helpers."""
def __init__(self, adapter: AgentA2AAdapter[Any]) -> None:
self.adapter = adapter
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
if context.context_id is None:
raise ValueError("A2A context id is required")
updater = TaskUpdater(event_queue, context.task_id or "", context.context_id)
await updater.cancel()
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
if context.message is None or context.context_id is None:
raise ValueError("A2A message and context id are required")
task = context.current_task
if task is None:
task = new_task_from_user_message(context.message)
await event_queue.enqueue_event(task)
updater = TaskUpdater(event_queue, task.id, context.context_id)
await updater.submit()
try:
await updater.start_work()
run = self.adapter.a2a_to_run(context.message, stream=True)
agent = await self.adapter.state.get_target()
# Demo-only key: the outer server must authenticate and authorize these protocol IDs for multi-user use.
session_id = f"a2a:{context.tenant}:{context.context_id}"
session = await self.adapter.state.get_or_create_session(session_id)
if not run["stream"]:
raise RuntimeError("This executor requires streaming run arguments.")
stream = agent.run( # pyright: ignore[reportCallIssue]
run["messages"],
session=session,
options=run["options"],
stream=run["stream"],
)
default_artifact_id = uuid.uuid4().hex
streamed_artifact_ids: set[str] = set()
async for update in stream:
parts = self.adapter.a2a_from_run(update)
if parts:
artifact_id = update.message_id or default_artifact_id
await updater.add_artifact(
parts=parts,
artifact_id=artifact_id,
append=True if artifact_id in streamed_artifact_ids else None,
)
streamed_artifact_ids.add(artifact_id)
final_response = await stream.get_final_response()
if not streamed_artifact_ids:
parts = self.adapter.a2a_from_run(final_response)
if parts:
await updater.update_status(
state=TaskState.TASK_STATE_WORKING,
message=updater.new_agent_message(parts),
)
await self.adapter.state.set_session(session_id, session)
await updater.complete()
except asyncio.CancelledError:
await updater.update_status(state=TaskState.TASK_STATE_CANCELED)
except Exception:
logger.exception("A2A agent execution failed.")
await updater.update_status(
state=TaskState.TASK_STATE_FAILED,
message=updater.new_agent_message([Part(text="Agent execution failed.")]),
)
La configuration du serveur crée l’adaptateur, génère son homologue natif AgentCard, puis combine l’exécuteur propre à l’application avec le gestionnaire de requêtes du SDK A2A :
if __name__ == "__main__":
flight_skill = InlineSkill(
frontmatter=SkillFrontmatter(
name="flight-booking",
description="Search and book flights across Europe.",
),
instructions="Help users search and book flights across Europe.",
)
hotel_skill = InlineSkill(
frontmatter=SkillFrontmatter(
name="hotel-booking",
description="Search and book hotels across Europe.",
),
instructions="Help users search and book hotels across Europe.",
)
agent = Agent(
client=OpenAIChatClient(),
name="Europe Travel Agent",
description="Helps users search and book flights and hotels across Europe.",
instructions="You are a helpful Europe Travel Agent.",
context_providers=[SkillsProvider([flight_skill, hotel_skill])],
)
state = AgentState(agent)
adapter = AgentA2AAdapter(
state,
version="1.0.0",
capabilities=AgentCapabilities(streaming=True),
supported_interfaces=[AgentInterface(url="http://localhost:9999/", protocol_binding="JSONRPC")],
)
public_agent_card = asyncio.run(adapter.get_card())
request_handler = DefaultRequestHandler(
agent_executor=AppAgentExecutor(adapter),
task_store=InMemoryTaskStore(),
agent_card=public_agent_card,
)
Créer un exécuteur A2A appartenant à l’application
Utilisez les assistances d’hébergement autonomes lorsque votre application a également besoin d’un contrôle direct sur la création de cartes :
pip install --pre agent-framework-hosting-a2a starlette uvicorn
Les assistants sont indépendants de tout framework :
-
a2a_to_runconvertit un A2AMessageen arguments d’exécution d’Agent Framework. -
a2a_from_runconvertit les réponses d’Agent Framework et les mises à jour en continu en valeurs A2APart.
Votre exécuteur sélectionne les clés de session et gère les transitions entre tâches, les files d’attente d’événements, les identifiants d’artefacts, les délimitations des messages et la distribution sortante.
a2a_from_run retourne une liste de parties plates afin que l’application puisse regrouper ces parties dans des messages ou artefacts A2A et appliquer des métadonnées au niveau du message.
La configuration d’hébergement prend également en charge les applications multi-protocole. Partagez la même cible d’agent et la même infrastructure AgentState sur les routes A2A, OpenAI Responses, Telegram et MCP, tandis que chaque point de terminaison du protocole conserve sa propre stratégie de conversion, d’autorisation et de clé de session. Cela permet aux clients d’atteindre un agent via différents protocoles en même temps sans créer de déploiement d’agent distinct pour chaque point de terminaison.
Composez les helpers dans un exécuteur natif du SDK A2A. Cet exemple crée et met à jour des tâches A2A, convertit le message entrant en une exécution Agent Framework, enregistre la session mise à jour AgentState une fois le flux terminé et publie les éléments renvoyés en tant qu’artefacts.
class AppAgentExecutor(AgentExecutor, Generic[AgentT]):
"""Native A2A SDK executor composed with Agent Framework conversion helpers."""
def __init__(self, state: AgentState[AgentT]) -> None:
self.state = state
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
if context.context_id is None:
raise ValueError("A2A context id is required")
updater = TaskUpdater(event_queue, context.task_id or "", context.context_id)
await updater.cancel()
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
if context.message is None or context.context_id is None:
raise ValueError("A2A message and context id are required")
task = context.current_task
if task is None:
task = new_task_from_user_message(context.message)
await event_queue.enqueue_event(task)
updater = TaskUpdater(event_queue, task.id, context.context_id)
await updater.submit()
try:
await updater.start_work()
run = a2a_to_run(context.message, stream=True)
agent = await self.state.get_target()
# Demo-only key: the outer server must authenticate and authorize these protocol IDs for multi-user use.
session_id = f"a2a:{context.tenant}:{context.context_id}"
session = await self.state.get_or_create_session(session_id)
if not run["stream"]:
raise RuntimeError("This executor requires streaming run arguments.")
stream = agent.run( # pyright: ignore[reportCallIssue]
run["messages"],
session=session,
options=run["options"],
stream=run["stream"],
)
default_artifact_id = uuid.uuid4().hex
streamed_artifact_ids: set[str] = set()
async for update in stream:
parts = a2a_from_run(update)
if parts:
artifact_id = update.message_id or default_artifact_id
await updater.add_artifact(
parts=parts,
artifact_id=artifact_id,
append=True if artifact_id in streamed_artifact_ids else None,
)
streamed_artifact_ids.add(artifact_id)
final_response = await stream.get_final_response()
if not streamed_artifact_ids:
parts = a2a_from_run(final_response)
if parts:
await updater.update_status(
state=TaskState.TASK_STATE_WORKING,
message=updater.new_agent_message(parts),
)
await self.state.set_session(session_id, session)
await updater.complete()
except CancelledError:
await updater.update_status(state=TaskState.TASK_STATE_CANCELED)
except Exception:
logger.exception("A2A agent execution failed.")
await updater.update_status(
state=TaskState.TASK_STATE_FAILED,
message=updater.new_agent_message([Part(text="Agent execution failed.")]),
)
L’exemple utilise Starlette et Uvicorn, mais les assistances ne sont pas liées à l’un ou l’autre. Utilisez votre framework applicatif ou un générateur d’applications du SDK A2A pour exposer la carte d’agent A2A et les routes JSON-RPC :
# Create the Agent Framework agent for the chosen type
agent_factory = AGENT_FACTORIES[args.agent_type]
agent = agent_factory(client)
state = AgentState(agent)
# Build the A2A server components
url = f"http://{args.host}:{args.port}/"
agent_card = AGENT_CARD_FACTORIES[args.agent_type](url)
executor = AppAgentExecutor(state)
task_store = InMemoryTaskStore()
request_handler = DefaultRequestHandler(
agent_executor=executor,
task_store=task_store,
agent_card=agent_card,
)
app = Starlette(
routes=[
*create_agent_card_routes(agent_card),
*create_jsonrpc_routes(request_handler, "/"),
]
)
Héberger un flux de travail avec un adaptateur
WorkflowA2AAdapter fournit la même limite de génération de cartes et de conversion pour un flux de travail ou WorkflowState. Il déduit les modes d’entrée et de sortie conservateurs des types déclarés du workflow, ou vous pouvez fournir des modes explicites pour une représentation spécifique à l’application.
Les fonctions utilitaires autonomes a2a_to_workflow_run et a2a_from_workflow_run assurent une conversion typée des entrées et sorties de workflow. L’adaptateur les expose sous forme de méthodes asynchrones a2a_to_run et synchrones a2a_from_run qui valident par rapport à ses modes de carte effectifs. La conversion d’entrée accepte un texte A2A, brut ou un élément de données pour le type d’entrée unique de l’exécuteur de démarrage du flux de travail, et la conversion de sortie associe les sorties publiques du flux de travail terminé à des éléments A2A natifs. Appel get_card avant la conversion de sortie validée lorsque l’adaptateur doit déduire les modes de sortie.
L’application reste responsable de l’exécuteur A2A natif ainsi que de la diffusion en continu de la progression, de l’état des tâches, des artefacts, des points de contrôle et de la poursuite avec intervention humaine. Les demandes d’entrée humaine en attente ne sont pas converties automatiquement. Par conséquent, l’hôte doit implémenter sa propre stratégie de continuation.
Sécuriser les sessions et l’état des tâches
A2AExecutor utilise l’ID de contexte A2A comme ID de session Agent Framework. Les exemples basés sur l’adaptateur et l’assistant combinent le locataire A2A et l’identifiant de contexte afin d’illustrer un mappage sélectionné par l’application. Dans chaque approche, un hôte de production doit authentifier l’appelant avant que la requête n’atteigne le gestionnaire de requêtes A2A, déterminer le locataire et le sujet à partir de cette identité de confiance, et autoriser tous les identifiants de tâche, de contexte, de continuation et d’annulation.
Important
Les stockages par défaut des tâches et de la configuration push du SDK A2A sont en mémoire et sont associés à un utilisateur selon son nom d’utilisateur. Pour un service multilocataire, utilisez un owner_resolver qui dérive la propriété du même locataire et du même sujet approuvé, et utilisez des magasins de tâches et de sessions durables lorsque les réplicas peuvent redémarrer ou effectuer un scale-out.
Pour obtenir des exemples complets de serveur d’assistance et de multi-agent, consultez les exemples d’hébergement A2A. Pour connaître les clients A2A et les fonctionnalités de protocole, consultez le service d’agent A2A.
Étapes suivantes
Aller plus loin :