Связь агента с агентом: внедрение протокола A2A в ваши проекты ADK

pip install google-adk

{
  "version": "1.0",
  "name": "Expense Reimbursement Agent",
  "description": "An agent that helps process expense reimbursements",
  "endpoint": "https://your-agent-endpoint.com/a2a",
  "authentication": {
    "type": "oauth2",
    "oauth2_url": "https://your-auth-server.com/oauth2"
  },
  "capabilities": {
    "streaming": true
  },
  "skills": [
    {
      "name": "process_expense",
      "description": "Process an expense reimbursement request",
      "tags": ["finance", "reimbursement"],
      "examples": [
        "I need to submit a receipt for $75.50 for office supplies",
        "Process my travel expenses from the New York conference"
      ]
    }
  ]
}

from google.adk import Agent
# Assuming you have placed the A2A server code in a 'a2a_lib' directory
from a2a_lib.server import A2AServer 
from a2a_lib.schema import AgentCard, AgentSkill

# Define your agent logic using ADK
class ExpenseAgent(Agent):
    def __init__(self):
        super().__init__(
            name="Expense Reimbursement Agent",
            description="An agent that helps process expense reimbursements"
        )

    def process_expense(self, amount, description, receipt_url=None):
        # Implement expense processing logic
        return {"status": "approved", "amount": amount}

# Create your agent instance
agent = ExpenseAgent()

# Create an agent card that describes your agent's capabilities
agent_card = AgentCard(
    name="Expense Reimbursement Agent",
    description="An agent that helps process expense reimbursements",
    version="1.0",
    skills=[
        AgentSkill(
            name="process_expense",
            description="Process an expense reimbursement request"
        )
    ]
)

# Create an A2A server for your agent
# Note: You might need to adapt the A2AServer initialization based on the specific sample code you integrated.
a2a_server = A2AServer(
    agent=agent,
    agent_card=agent_card,
    host="0.0.0.0",
    port=8000
)

# Start the server
a2a_server.start()

# Assuming you have placed the A2A client code in a 'a2a_lib' directory
from a2a_lib.client import A2AClient
from a2a_lib.schema import TaskRequest

# Create an A2A client
a2a_client = A2AClient()

# Discover an agent by its URL
agent_card = a2a_client.discover("https://another-agent.com/.well-known/agent.json")

# Create a task request
task_request = TaskRequest(
    skill="analyze_data",
    inputs={"data_url": "https://example.com/data.csv"}
)

# Send a task to the agent
task = a2a_client.create_task(
    agent_url=agent_card.endpoint,
    task_request=task_request
)

# Get the task result (blocking until complete)
result = a2a_client.wait_for_task_result(task.id)
print(result)

# Alternatively, you can poll for the task status
# task_status = a2a_client.get_task(agent_url=agent_card.endpoint, task_id=task.id)
# print(f"Task status: {task_status.state}")

from google.adk import Agent
# Assuming you have placed the A2A code in a 'a2a_lib' directory
from a2a_lib.server import A2AServer
from a2a_lib.client import A2AClient
from a2a_lib.schema import AgentCard, AgentSkill, TaskRequest

# Define the main travel agent
class TravelAgent(Agent):
    def __init__(self):
        super().__init__(
            name="Travel Planning Agent",
            description="An agent that helps plan travel itineraries"
        )
        self.a2a_client = A2AClient()

    def plan_trip(self, destination, dates, budget):
        # Discover flight booking agent
        flight_agent_card = self.a2a_client.discover("https://flight-agent-example.com/.well-known/agent.json")

        # Create flight task request
        flight_task_request = TaskRequest(
            skill="find_flights",
            inputs={"destination": destination, "dates": dates, "budget": budget}
        )

        # Send task to flight booking agent
        flight_task = self.a2a_client.create_task(
            agent_url=flight_agent_card.endpoint,
            task_request=flight_task_request
        )

        # Discover hotel booking agent
        hotel_agent_card = self.a2a_client.discover("https://hotel-agent-example.com/.well-known/agent.json")

        # Create hotel task request
        hotel_task_request = TaskRequest(
            skill="find_hotels",
            inputs={"destination": destination, "dates": dates, "budget": budget}
        )

        # Send task to hotel booking agent
        hotel_task = self.a2a_client.create_task(
            agent_url=hotel_agent_card.endpoint,
            task_request=hotel_task_request
        )

        # Get results from both agents (blocking until complete)
        flight_result = self.a2a_client.wait_for_task_result(flight_task.id)
        hotel_result = self.a2a_client.wait_for_task_result(hotel_task.id)

        # Combine results into a complete travel plan
        return {
            "destination": destination,
            "dates": dates,
            "flights": flight_result.output["options"],
            "hotels": hotel_result.output["options"],
            "total_cost": flight_result.output["best_price"] + hotel_result.output["best_price"]
        }

# Create the agent card for our travel agent
travel_agent_card = AgentCard(
    name="Travel Planning Agent",
    description="An agent that helps plan travel itineraries",
    version="1.0",
    skills=[
        AgentSkill(
            name="plan_trip",
            description="Plan a complete trip including flights and hotels"
        )
    ]
)

# Create and start the A2A server
agent = TravelAgent()
# Note: You might need to adapt the A2AServer initialization based on the specific sample code you integrated.
a2a_server = A2AServer(
    agent=agent,
    agent_card=travel_agent_card,
    host="0.0.0.0",
    port=8000
)
a2a_server.start()