Communicating Between uAgents and Google ADK: A Step-by-Step Guide

1M ago•

bullish:

bearish:

In the world of decentralized and AI-driven applications, integrating agent-based systems with powerful AI frameworks can unlock new possibilities. This post explores how to enable communication between uAgents (from Fetch.ai) and Google ADK (Agent Development Kit), two platforms designed to build intelligent agents. We’ll dive into what each framework offers, how they can work together, and provide a practical example of setting up REST-based communication for seamless interaction.

What is uAgents?

uAgents is a Python library by Fetch.ai that enables developers to create autonomous agents capable of communicating over a decentralized network. These agents can:

Send and receive messages using unique addresses.
Expose REST APIs for external interactions.
Perform tasks like fetching data, executing logic, or interacting with blockchains.

Key features:

Agent Addresses: Each agent has a unique identifier (e.g., agent1q29anz...) for peer-to-peer messaging.
REST Endpoints: Agents can host HTTP endpoints (e.g., /coin) to handle requests.
Event Handlers: Support for on_event(), on_interval(), on_message(), on_rest_get(), on_rest_post() to manage workflows.

uAgents is ideal for building decentralized systems where agents need to collaborate without central control.

What is Google ADK?

Google ADK (Agent Development Kit) is a framework for creating conversational AI agents powered by Google’s generative models (e.g., Gemini). It simplifies building agents that:

Process natural language queries.
Execute functions via tools (e.g., FunctionTool).
Maintain session state for context-aware interactions.

Key features:

LlmAgent: A core component for defining agents with instructions and tools.
FunctionTool: Allows agents to call custom functions (e.g., API requests) based on user input.
Session Management: Tracks conversation state using InMemorySessionService.

ADK excels at creating user-facing AI agents that integrate with external services.

Why Integrate uAgents and Google ADK?

Combining uAgents and Google ADK lets you leverage:

uAgents’ Decentralized Communication: For agent-to-agent messaging or blockchain interactions.
ADK’s Conversational Power: For natural language understanding and user interaction.
Hybrid Workflows: Use uAgents for backend logic (e.g., fetching crypto prices) and ADK for frontend query handling.

A common use case is a cryptocurrency information bot where:

The ADK agent handles user queries like “Get details for Bitcoin.”
The uAgent fetches data from an API (e.g., CoinGecko) and responds via REST.

How to Communicate: uAgents and Google ADK

To make uAgents and Google ADK communicate, we’ll use REST APIs as the bridge, as it’s reliable and avoids complex protocol mismatches. Here’s the approach:

uAgent Setup: Create a uAgent that exposes a REST endpoint (e.g., /coin) to provide data.
ADK Agent Setup: Build an ADK agent that calls the uAgent’s REST endpoint using a FunctionTool.
Data Flow:

ADK agent receives a user query (e.g., “Details for Bitcoin”).
ADK sends a POST request to the uAgent’s endpoint.
uAgent processes the request, fetches data, and returns a JSON response.
ADK formats the response for the user.

Below is a practical example of a cryptocurrency information system.

Example: Cryptocurrency Information Bot

Let’s build a system where a Google ADK agent queries a uAgent for cryptocurrency details (e.g., price, market cap, description). The uAgent fetches data from CoinGecko, and the ADK agent presents it to the user.

Prerequisites

Python 3.8+
Virtual environment:

python3 -m venv .venv source 
.venv/bin/activate

Install dependencies:

pip install google-adk==0.1.0 uagents requests==2.31.0 python-dotenv==1.0.1

Google API Key from Google AI Studio.

.env file:

GOOGLE_API_KEY=your-gemini-api-key

Step 1: uAgent Implementation

Create coin_info_agent.py to fetch crypto data and expose a REST endpoint.

import time
import logging
import requests
from typing import Any, Dict
from uagents import Agent, Context, Model

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger("CoinInfoAgent")

# Define models
class CoinRequest(Model):
    coin_id: str

class CoinResponse(Model):
    status: str
    coin_id: str = None
    name: str = None
    symbol: str = None
    market_data: Dict = None
    description: str = None
    details: Dict = None
    timestamp: int = None
    error_message: str = None
    agent_address: str = None

# Create uAgent
coin_agent = Agent(
    name="coin_info_agent",
    port=8003,
    seed="coin_agent_secret_phrase",
    endpoint=["https://127.0.0.1:8003/submit"]
)

COINGECKO_API = "https://api.coingecko.com/api/v3/coins/"

# POST endpoint
@coin_agent.on_rest_post("/coin", CoinRequest, CoinResponse)
async def handle_coin_post(ctx: Context, req: CoinRequest) -> CoinResponse:
    logger.info("Received POST request for coin data")
    coin_id = req.coin_id.lower()
    symbol_to_id = {
        "btc": "bitcoin",
        "eth": "ethereum",
        "ada": "cardano"
    }
    coin_id = symbol_to_id.get(coin_id, coin_id)
    
    try:
        response = requests.get(
            f"{COINGECKO_API}{coin_id}?localization=false&tickers=false&community_data=false&developer_data=false",
            timeout=5
        )
        response.raise_for_status()
        coin_data = response.json()
        
        description = coin_data.get("description", {}).get("en", "No description available")
        return CoinResponse(
            status="success",
            coin_id=coin_data.get("id"),
            name=coin_data.get("name", "Unknown"),
            symbol=coin_data.get("symbol", "N/A").upper(),
            market_data={
                "current_price_usd": coin_data.get("market_data", {}).get("current_price", {}).get("usd"),
                "market_cap_usd": coin_data.get("market_data", {}).get("market_cap", {}).get("usd"),
                "24h_volume_usd": coin_data.get("market_data", {}).get("total_volume", {}).get("usd"),
                "24h_price_change_percentage": coin_data.get("market_data", {}).get("price_change_percentage_24h")
            },
            description=description[:500] + "..." if len(description) > 500 else description,
            details={
                "last_updated": coin_data.get("last_updated", "Unknown"),
                "homepage": coin_data.get("links", {}).get("homepage", [""])[0]
            },
            timestamp=int(time.time()),
            agent_address=ctx.agent.address
        )
    except Exception as e:
        logger.error(f"Failed to fetch coin data: {str(e)}")
        return CoinResponse(
            status="error",
            error_message=str(e),
            timestamp=int(time.time()),
            agent_address=ctx.agent.address
        )

# Health endpoint
@coin_agent.on_rest_get("/health", CoinResponse)
async def handle_health(ctx: Context) -> Dict[str, Any]:
    logger.info("Received health check request")
    return {
        "status": "healthy",
        "timestamp": int(time.time()),
        "agent_address": ctx.agent.address
    }

# Startup event
@coin_agent.on_event("startup")
async def startup(ctx: Context):
    logger.info(f"Coin agent started with address: {ctx.agent.address}")

if __name__ == "__main__":
    coin_agent.run()

What it does:

Defines CoinRequest (input: coin_id) and CoinResponse (output: coin details).
Listens on https://127.0.0.1:8003/coin for POST requests.
Fetches data from CoinGecko and returns formatted JSON.

Step 2: Google ADK Agent Implementation

Create agent.py to query the uAgent and format the response.

import asyncio
import logging
import os
import requests
from dotenv import load_dotenv
from google.adk.agents import LlmAgent
from google.adk.sessions import InMemorySessionService
from google.adk.tools.function_tool import FunctionTool

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger("ADKCoinAgent")

load_dotenv()

GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
if not GOOGLE_API_KEY:
    raise ValueError("GOOGLE_API_KEY must be set in .env")

COIN_API_URL = "https://127.0.0.1:8003/coin"

async def fetch_coin_details(coin: str) -> str:
    logger.info(f"Querying details for {coin}")
    try:
        response = requests.post(
            COIN_API_URL,
            json={"coin_id": coin.lower()},
            timeout=5
        )
        response.raise_for_status()
        data = response.json()
        
        if data["status"] == "success":
            return (
                f"**Details for {data['name']} ({data['symbol']})**\n"
                f"- **Coin ID**: {data['coin_id']}\n"
                f"- **Name**: {data['name']}\n"
                f"- **Symbol**: {data['symbol']}\n"
                f"- **Market Data**:\n"
                f"  - Current Price: ${data['market_data']['current_price_usd']:.2f} USD\n"
                f"  - Market Cap: ${data['market_data']['market_cap_usd']:,.0f} USD\n"
                f"  - 24h Volume: ${data['market_data']['24h_volume_usd']:,.0f} USD\n"
                f"  - 24h Price Change: {data['market_data']['24h_price_change_percentage']:.2f}%\n"
                f"- **Description**: {data['description']}\n"
                f"- **Additional Info**:\n"
                f"  - Last Updated: {data['details']['last_updated']}\n"
                f"  - Homepage: {data['details']['homepage']}\n"
                f"- **Agent Address**: {data['agent_address']}"
            )
        return f"Error: {data.get('error_message', 'Unknown error')}"
    except Exception as e:
        logger.error(f"Failed to fetch coin details: {str(e)}")
        return f"Error contacting coin service: {str(e)}"

coin_details_tool = FunctionTool(func=fetch_coin_details)

root_agent = LlmAgent(
    model="gemini-2.0-flash-exp",
    name="coin_agent",
    instruction="Fetch cryptocurrency details using the fetch_coin_details tool when asked (e.g., 'Details for Bitcoin').",
    tools=[coin_details_tool]
)

if __name__ == "__main__":
    logger.info("Starting ADK agent")
    session_service = InMemorySessionService()
    try:
        asyncio.run(root_agent.start(session_service=session_service))
    except Exception as e:
        logger.error(f"Failed to start agent: {str(e)}")
        exit(1)

What it does:

Uses LlmAgent with a FunctionTool to call the uAgent’s /coin endpoint.
Formats the response with all requested fields (price, description, etc.).
Runs without Flask, accepting dynamic queries via the ADK UI.

Step 3: Run the System

Start uAgent:

(.venv) % python3 coin_info_agent.py

Logs: INFO:CoinInfoAgent:Coin agent started with address: agent1q29anz...

Agent Decorators:
The uAgents framework provides several decorators to handle different types of events and interactions:

on_event(): Triggers when specific agent events occur (e.g., startup, shutdown).
on_interval(): Executes a function at regular time intervals.
on_message(): Handles incoming messages sent to the agent’s address.
on_rest_get(): Defines a REST GET endpoint for the agent.
on_rest_post(): Defines a REST POST endpoint for the agent.

2 Launch ADK UI:

(.venv) % adk web

Open https://localhost:8000/dev-ui?app=coin_app.

Enter: “Details for Bitcoin”.

Step 4: Expected Output

In the ADK UI, you’ll see:

**Details for Bitcoin (BTC)**
- **Coin ID**: bitcoin
- **Name**: Bitcoin
- **Symbol**: BTC
- **Market Data**:
  - Current Price: $84,699.00 USD
  - Market Cap: $1,678,901,234,567 USD
  - 24h Volume: $45,678,901,234 USD
  - 24h Price Change: 2.34%
- **Description**: Bitcoin is a decentralized digital currency...
- **Additional Info**:
  - Last Updated: 2025-04-13T...
  - Homepage: https://bitcoin.org
- **Agent Address**: agent1q29anz...

Advanced Communication: uAgents Messaging

While this example uses REST for simplicity, you can extend it to use uAgents’ messaging protocol:

uAgent to uAgent: Send messages using ctx.send to the uAgent’s address (e.g., agent1q29anz...).
ADK as uAgent: Wrap the ADK agent in a uAgent to send/receive messages.
Challenges: Requires syncing ADK’s async execution with uAgents’ event loop and handling timeouts.

Example snippet for uAgent messaging:

from uagents import Agent, Context, Model
class CoinRequest(Model):
    coin_id: str
    reply_url: str
coin_agent = Agent(name="coin_agent", port=8003)
@coin_agent.on_message(model=CoinRequest)
async def handle_message(ctx: Context, sender: str, msg: CoinRequest):
    # Process request and POST to msg.reply_url
    pass

This is more complex and may require additional synchronization, so REST is often preferred for ADK-uAgent integration.

Best Practices

Error Handling: Always include try-except blocks for API calls and validate responses.
Timeouts: Set reasonable timeouts (e.g., 5 seconds) for REST requests.
Logging: Use detailed logs to debug communication failures.
Environment Variables: Store sensitive keys (e.g., GOOGLE_API_KEY) in .env.
Port Management: Ensure ports (e.g., 8003) are free before starting agents.

Conclusion

Integrating uAgents and Google ADK combines the strengths of decentralized agent systems and conversational AI. By using REST APIs, you can create robust communication channels that are easy to implement and scale. The cryptocurrency bot example demonstrates how to fetch and present data seamlessly, but the pattern applies to any domain — finance, IoT, or beyond.

Try building your own uAgent-Google ADK system, and explore uAgents’ messaging for more decentralized use cases. Happy coding!

📧 Contact

For any inquiries or feedback, feel free to reach out to us at gautammanak1@gmail.com.

Communicating Between uAgents and Google ADK: A Step-by-Step Guide was originally published in Fetch.ai on Medium, where people are continuing the conversation by highlighting and responding to this story.

1M ago•

Fetch.AI

bullish:

bearish: