Not
Bu sayfaya erişim yetkilendirme gerektiriyor. Oturum açmayı veya dizinleri değiştirmeyi deneyebilirsiniz.
Bu sayfaya erişim yetkilendirme gerektiriyor. Dizinleri değiştirmeyi deneyebilirsiniz.
ASP.NET Core aracılığıyla bir Agent Framework aracısını kullanıma açmak için .NET A2A barındırma paketlerini kullanın. Paket kurulumu ve eksiksiz bir sunucu örneği için A2A tümleştirmesi bölümüne bakın.
Go provider/a2aprovider paketini resmi A2A Go sunucu işleyicileriyle kullanın. Eksiksiz bir sunucu örneği için bkz. A2A tümleştirmesi .
Agent Framework, resmi A2A SDK'sı aracılığıyla aracıları ve iş akışlarını barındırmak için iki Python paketi sağlar:
| Package | Tümleştirme modeli | Şu durumlarda kullanın: |
|---|---|---|
agent-framework-a2a |
İstekleri dönüştüren, bir ajan çalıştıran ve A2A görev olayları ile artefaktlarını yayımlayan, belirli kabulleri olan bir A2AExecutor |
Standart Agent Framework-A2A davranışını istiyorsunuz ve yalnızca A2A SDK sunucusunu derlemeniz gerekiyor. |
agent-framework-hosting-a2a |
Uygulamaya ait bir yürütücü için artımlı yapı taşları. Temel aracı veya iş akışı dönüştürücüleriyle başlayın; isteğe bağlı olarak, bu dönüştürücüleri temel alarak yerel kart oluşturma ve mod doğrulaması ekleyen AgentA2AAdapter veya WorkflowA2AAdapter kullanın. |
Uygulamanızın oturum eşlemesine, görev geçişlerine, olay teslimine, yapıt sınırlarına, çıkış dönüşümüne veya çok protokollü bir konağa sahip olması gerekir. |
Her iki paket de yerel A2A SDK türlerini ve sunucu bileşenlerini kullanır. Uygulamanız istek işleyicisini, görev deposunu, yolları veya SDK uygulama oluşturucusunu, kimlik doğrulamasını ve dağıtımını sağlar.
agent-framework-hosting-a2a ile uygulama, ajan kartını doğrudan oluşturabilir veya bir bağdaştırıcının bunu oluşturmasına izin verebilir.
Fikir sahibi A2A yürütücüsü kullanma
Yerleşik sunucu bağdaştırıcısı yaşam döngünüze uyuyorsa agent-framework-a2a yükleyin:
pip install --pre agent-framework-a2a starlette uvicorn
A2AExecutor A2A SDK'sının AgentExecutoruygulamasını uygular. A2A istek bağlamından kullanıcı girişini okur, A2A bağlam kimliğinden bir Agent Framework oturumu oluşturur, aracıyı akış veya akış dışı modda çalıştırır, desteklenen çıkış içeriğini dönüştürür ve SDK'nın TaskUpdateraracılığıyla görev durumu ve yapıt olaylarını yayımlar.
Bunu A2A SDK’sının DefaultRequestHandler, görev deposu, ajan kartı ve Starlette uygulaması veya desteklenen başka bir sunucu tümleştirmesi ile oluşturun. Akışı A2AExecutor(agent, stream=True) ile yapılandırın, kararlı ajan çalıştırma seçeneklerini run_kwargs aracılığıyla iletin ya da farklı bir çıkış eşlemesine ihtiyaç duyduğunuzda A2AExecutor sınıfından türeyip handle_events öğesini geçersiz kılın.
A2AExecutor bir A2A uç noktasıyla sınırlıdır ve kendi A2A yürütmesini ve oturum eşlemesini doğrudan yönetir. Aynı aracının tek bir uygulamada çeşitli protokoller aracılığıyla kullanılabilir olması gerektiğinde barındırma paketlerini kullanın.
Sunucu kurulumunun tamamı için bkz. A2A üzerinden Bir Agent Framework aracısını kullanıma sunma.
Uygulamaya ait bir yürütücüde bağdaştırıcı kullanma
Uygulamanız yerel A2A yürütücüsüne sahipse ancak Agent Framework’ün genel kartı oluşturmasını ve dönüşümleri doğrulamasını istiyorsa, barındırma paketini yükleyin:
pip install --pre agent-framework-hosting-a2a starlette uvicorn
AgentA2AAdapter, bir aracı veya AgentState kabul eder. Eşzamansız get_card yöntemi, açık adını ve açıklamasını elde eder, varsayılan olarak muhafazakâr metin modlarını kullanır ve Agent Framework SkillsProvider örneklerinden yerel A2A becerilerini çıkarabilir. Sunucu özellikleri ve desteklenen arabirimler, aracının run yöntemi yerine uygulama uç noktasını tanımladığından açık kalır.
Bağdaştırıcı, varsayılan olarak değerleri yapılandırılmış kart modlarına göre doğrulayan a2a_to_run ve a2a_from_run yöntemlerini sunar. Uygulama hâlâ A2A yürütücüsünü, görev yaşam döngüsünü, olay kuyruğunu, artefakt sınırlarını, oturum ilkesini, kimlik doğrulamayı, yönlendirmeleri ve dağıtımı yönetir.
Bu yürütücü, gelen dönüşüm, ajan durumu ve giden dönüşüm için tek bir bağdaştırıcı kullanır:
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.")]),
)
Sunucu yapılandırması bağdaştırıcıyı oluşturur, kendi yerel AgentCard bileşenini üretir ve uygulamaya ait yürütücüyü A2A SDK istek işleyicisiyle birleştirir:
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,
)
Uygulamaya ait bir A2A çalıştırıcısı oluşturun
Uygulamanızın kart oluşturma üzerinde doğrudan denetime de ihtiyacı olduğunda tek başına barındırma yardımcılarını kullanın:
pip install --pre agent-framework-hosting-a2a starlette uvicorn
Yardımcılar çerçeveden bağımsızdır:
-
a2a_to_run, bir A2AMessageöğesini Agent Framework çalıştırma argümanlarına dönüştürür. -
a2a_from_runAgent Framework yanıtlarını ve akış güncelleştirmelerini A2APartdeğerlerine dönüştürür.
Yürütücünüz oturum anahtarlarını seçer ve görev geçişlerinin, olay kuyruklarının, yapıt kimliklerinin, ileti sınırlarının ve giden teslimin sahibidir.
a2a_from_run uygulamanın bu bölümleri A2A iletileri veya yapıtları halinde gruplandırabilmesi ve ileti düzeyinde meta veriler uygulayabilmesi için düz bir parça listesi döndürür.
Barındırma kurulumu, çok protokollü uygulamaları da destekler. A2A, OpenAI Responses, Telegram ve MCP rotaları arasında aynı aracı hedefini ve AgentState altyapıyı paylaşırken, her protokol uç noktası kendi dönüşüm, yetkilendirme ve oturum anahtarı ilkesini korur. Bu, istemcilerin her uç nokta için ayrı bir aracı dağıtımı oluşturmadan aynı anda farklı protokoller aracılığıyla bir aracıya ulaşmasını sağlar.
Yardımcıları yerel bir A2A SDK yürütücüsunda oluşturma. Bu örnek A2A görevlerini oluşturur ve güncelleştirir, gelen iletiyi bir Agent Framework çalıştırmasına dönüştürür, akış tamamlandıktan sonra güncelleştirilmiş AgentState oturumu kalıcı hale getirerek döndürülen bölümleri yapıt olarak yayımlar.
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.")]),
)
Örnekte Starlette ve Uvicorn kullanılmıştır, ancak yardımcılar da buna bağlı değildir. A2A aracı kartına ve JSON-RPC yollarına hizmet vermek için uygulama çerçevenizi veya A2A SDK uygulama oluşturucunuzu kullanın:
# 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, "/"),
]
)
Bir iş akışını adaptör ile barındırın
WorkflowA2AAdapter bir iş akışı veya WorkflowStateiçin aynı kart oluşturma ve dönüştürme sınırını sağlar. İş akışının bildirilen türlerinden muhafazakar giriş ve çıkış modlarını çıkarsar veya uygulamaya özgü bir gösterim için açık modlar sağlayabilirsiniz.
Bağımsız a2a_to_workflow_run ve a2a_from_workflow_run yardımcıları, tiplenmiş iş akışı girdisi ve çıktısı dönüşümü sağlar. Bağdaştırıcı, bunları geçerli kart modlarına göre doğrulayan zaman uyumsuz a2a_to_run ve zaman uyumlu a2a_from_run yöntemleri olarak sunar. Girdi dönüştürme, iş akışındaki tek başlangıç yürütücüsü girdi türü için tek bir A2A metin, ham veya veri parçasını kabul eder ve çıktı dönüştürme, tamamlanmış genel iş akışı çıktılarını yerel A2A parçalarına eşler. Bağdaştırıcının çıkış modlarını çıkarsaması gerektiğinde, doğrulanmış çıkış dönüşümünden önce get_card öğesini çağırın.
Uygulama, yerel A2A yürütücüsünden ve ilerleme akışı, görev durumu, yapıtlar, denetim noktaları ile insanın devrede olduğu devam sürecinden sorumlu olmaya devam eder. İnsan girdisi gerektiren bekleyen istekler otomatik olarak işlenmez; bu nedenle ana bilgisayar kendi devam ettirme ilkesini uygulamalıdır.
Güvenli oturumlar ve görev durumu
A2AExecutor Aracı Çerçevesi oturum kimliği olarak A2A bağlam kimliğini kullanır. Adaptör tabanlı ve yardımcı tabanlı örnekler, uygulama tarafından seçilen bir eşlemeyi göstermek için A2A kiracısını ve bağlam kimliğini birleştirir. Her yaklaşımda, üretim ana bilgisayarının A2A istek işleyicisine ulaşmadan önce çağıranın kimliğini doğrulaması, kiracıyı ve konuyu bu güvenilir kimlikten türetmesi ve tüm görev, bağlam, devamlılık ve iptal kimliklerini yetkilendirmesi gerekir.
Important
A2A SDK’sinin varsayılan görev ve itme yapılandırması depoları bellek içidir ve kapsam sahipliği kullanıcı adına göredir. Çok kiracılı bir hizmet için, sahipliği aynı güvenilir kiracıdan ve özneden türeten bir owner_resolver kullanın; ayrıca replikaların yeniden başlatılabildiği veya ölçek genişletilebildiği durumlarda dayanıklı görev ve oturum depoları kullanın.
Tam bir yardımcı tabanlı sunucu ve çok aracılı örnekler için bkz. A2A barındırma örnekleri. A2A istemcileri ve protokol özellikleri için bkz. A2A tümleştirmesi.
Sonraki Adımlar
Daha derine gidin: