# Install Google ADK
pip install google-adk
 
# Install the A2A extension
pip install google-adk-a2a

# agent.py
from google.adk import Agent
from google.adk.extensions.a2a import A2AServer
from google.adk.tool import ToolContext
 
# Create your ADK agent
agent = Agent(
    name="Google ADK Weather Agent",
    description="Provides weather information for any location",
    llm="gemini-pro"
)
 
# Define the agent's capabilities
@agent.register_tool
def get_weather(location: str) -> dict:
    """Get current weather for a specific location."""
    # Implementation details...
    return {
        "temperature": 22,
        "condition": "Sunny",
        "humidity": 65,
        "wind": "5 mph"
    }
 
# Create the A2A server wrapping the ADK agent
a2a_server = A2AServer(
    agent=agent,
    agent_card={
        "name": "Weather Agent",
        "description": "Provides weather information for locations worldwide",
        "version": "1.0.0",
        "authentication": {
            "schemes": ["Bearer"]
        },
        "capabilities": {
            "streaming": True
        },
        "defaultInputModes": ["text/plain"],
        "defaultOutputModes": ["text/markdown", "application/json"],
        "skills": [
            {
                "id": "weather-lookup",
                "name": "Weather Lookup",
                "description": "Get current weather conditions for a location",
                "tags": ["weather", "forecast"],
                "examples": [
                    "What's the weather in Paris?",
                    "Is it raining in Tokyo?"
                ]
            }
        ]
    }
)
 
# Start the server
if __name__ == "__main__":
    a2a_server.serve("0.0.0.0", 8080)

# Deploy to Google Cloud Run
gcloud run deploy weather-agent \
  --source . \
  --allow-unauthenticated \
  --port 8080

# client.py
from google.adk import Agent
from google.adk.extensions.a2a import A2AClient
from google.adk.tool import ToolContext
 
# Create an ADK agent
agent = Agent(
    name="Meta Agent",
    description="A meta-agent that coordinates with other A2A agents",
    llm="gemini-pro"
)
 
# Create an A2A client
weather_client = A2AClient(
    url="https://weather-agent-example.com/a2a",
    auth_token="your-auth-token"
)
 
# Register the A2A agent as a tool
@agent.register_tool
async def ask_weather_agent(ctx: ToolContext, query: str) -> str:
    """Ask the weather agent for information."""
    # Create a task with the A2A agent
    task = await weather_client.create_task(
        message={
            "parts": [
                {
                    "mimeType": "text/plain",
                    "data": query
                }
            ]
        }
    )
    
    # Wait for completion
    completed_task = await weather_client.wait_for_task(task["taskId"])
    
    # Extract and return the weather information
    if completed_task["status"] == "Completed" and completed_task["artifacts"]:
        artifact = completed_task["artifacts"][0]
        return artifact["parts"][0]["data"]
    else:
        return "Failed to get weather information"
 
# Use the agent
def main():
    response = agent.chat("What's the weather like in San Francisco?")
    print(response)
 
if __name__ == "__main__":
    main()

from google.adk.extensions.a2a import A2AClient, StreamingHandler
 
class MyStreamHandler(StreamingHandler):
    def on_progress(self, content):
        print(f"Progress: {content}")
    
    def on_complete(self, task):
        print("Task completed!")
    
    def on_error(self, error):
        print(f"Error: {error}")
 
async def stream_from_agent():
    client = A2AClient(url="https://example-agent.com/a2a")
    
    # Create a streaming task
    await client.create_streaming_task(
        message={
            "parts": [
                {
                    "mimeType": "text/plain",
                    "data": "Generate a detailed report on climate change"
                }
            ]
        },
        handler=MyStreamHandler()
    )

from google.adk import Agent
from google.adk.extensions.a2a import A2AClient
from google.adk.extensions.a2a.registry import AgentRegistry
 
# Create a registry of A2A agents
registry = AgentRegistry()
 
# Register external A2A agents
registry.register_agent(
    "weather",
    A2AClient(url="https://weather-agent.example.com/a2a")
)
registry.register_agent(
    "travel",
    A2AClient(url="https://travel-agent.example.com/a2a")
)
registry.register_agent(
    "news",
    A2AClient(url="https://news-agent.example.com/a2a")
)
 
# Create the router agent
router = Agent(
    name="Agent Router",
    description="Routes queries to specialized agents",
    llm="gemini-pro"
)
 
@router.register_tool
async def route_query(query: str) -> str:
    """Route the query to the appropriate agent."""
    # Determine which agent to use (implementation omitted)
    agent_type = determine_agent_type(query)
    
    # Get the client for the appropriate agent
    client = registry.get_agent(agent_type)
    
    # Create and process the task
    task = await client.create_task(
        message={
            "parts": [
                {
                    "mimeType": "text/plain",
                    "data": query
                }
            ]
        }
    )
    
    completed_task = await client.wait_for_task(task["taskId"])
    
    # Return the result
    if completed_task["status"] == "Completed" and completed_task["artifacts"]:
        return completed_task["artifacts"][0]["parts"][0]["data"]
    else:
        return f"Failed to get a response from the {agent_type} agent"

from google.adk.extensions.a2a import A2AServer
from google.adk.extensions.a2a.auth import BearerTokenAuthenticator, OAuthAuthenticator
 
# Bearer token authentication
bearer_auth = BearerTokenAuthenticator(
    tokens=["token1", "token2"],  # List of valid tokens
    token_validator=lambda token: validate_token(token)  # Custom validation
)
 
# OAuth authentication
oauth_auth = OAuthAuthenticator(
    issuer="https://accounts.google.com",
    audience="your-audience-id"
)
 
# Create the server with authentication
a2a_server = A2AServer(
    agent=agent,
    agent_card=agent_card,
    authenticator=bearer_auth  # or oauth_auth
)

# Deploy as a Cloud Run service
gcloud run deploy a2a-agent \
  --source . \
  --region us-central1 \
  --memory 1Gi \
  --cpu 1 \
  --allow-unauthenticated

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: a2a-agent
spec:
  replicas: 3
  selector:
    matchLabels:
      app: a2a-agent
  template:
    metadata:
      labels:
        app: a2a-agent
    spec:
      containers:
      - name: a2a-agent
        image: gcr.io/your-project/a2a-agent:latest
        ports:
        - containerPort: 8080
        env:
        - name: GOOGLE_APPLICATION_CREDENTIALS
          value: /secrets/service-account.json
        volumeMounts:
        - name: google-creds
          mountPath: /secrets
          readOnly: true
      volumes:
      - name: google-creds
        secret:
          secretName: google-service-account
---
apiVersion: v1
kind: Service
metadata:
  name: a2a-agent-service
spec:
  selector:
    app: a2a-agent
  ports:
  - port: 80
    targetPort: 8080
  type: LoadBalancer