Agent-Based Financial Intelligence: A Practical Implementation Guide
Agent Development Kit (ADK) is an open-sourced python library designed to empower developers to build, manage, evaluate, and deploy AI-powered agents. It provides a robust and flexible environment for creating both conversational and non-conversational agents capable of handling complex tasks and workflows.
Before getting started, make sure you have the following:
- Python 3.8+ installed on your machine
- Google Cloud account with billing enabled (for the Cloud Run deployment)
- gcloud CLI installed and configured
- Alpha Vantage API key — Register for a free API key at Alpha Vantage
Installation
To get started, install the Google ADK in your Python virtual environment using the commands below.
# Create your virtual environment
python3 -m venv .venv# Install the Google ADK
pip install google-adkInstall
With our installation complete, let’s create our Stock Market AI agent using the Google ADK Framework. Make sure you have your Alpha Vantage API key ready, as we’ll need it for the stock data functions.
Creating our stock market agent
Set your project directory this way using the command below
mkdir -p Stock_agent && touch Stock_agent/__init__.py \
Stock_agent/agent.py \
Stock_agent/.env
Stock_agent/
├── __init__.py
├── agent.py
└── .env
Import your agent module in your __init__.py file:
from . import agent
We’ll define our agent and tools in our agent.py file. Tools in this context are referred to as helper functions that help our agent accomplish tasks. For our stock market agent, we’ll have 2 functions (tools) called get_market_movers and get_stock_details. They both use the Alpha Vantage API to fetch stock market information and return data used by the AI agent to answer users’ queries.
First, make sure your .env file contains your Alpha Vantage API key:
ALPHA_VANTAGE_API_KEY=your_api_key_here
GOOGLE_API_KEY=yourgemini_api_key
Now, let’s create the agent code:
import os
import datetime
import requests
from typing import Dict, Any, List
from dotenv import load_dotenv
from google.adk.agents import Agent# Load environment variables
load_dotenv()
# Get API key from environment
ALPHA_VANTAGE_API_KEY = os.getenv("ALPHA_VANTAGE_API_KEY")
def get_market_movers(limit: int = 10) -> Dict[str, Any]:
"""Retrieves the top gainers and losers from the US stock market at the current time.
Args:
limit (int): Number of stocks to return for each category (gainers/losers).
Returns:
dict: Contains status and lists of top gainers and losers with their performance data.
"""
try:
# Call Alpha Vantage API for top gainers
gainers_url = f"https://www.alphavantage.co/query?function=TOP_GAINERS_LOSERS&apikey={ALPHA_VANTAGE_API_KEY}"
response = requests.get(gainers_url)
if response.status_code != 200:
return {
"status": "error",
"error_message": f"API request failed with status code {response.status_code}"
}
data = response.json()
# Process top gainers
gainers = []
if "top_gainers" in data:
for stock in data["top_gainers"][:limit]:
gainers.append({
"symbol": stock.get("ticker", ""),
"name": stock.get("name", ""),
"price": float(stock.get("price", "0").replace("$", "")),
"change": float(stock.get("change_amount", "0").replace("$", "")),
"change_percent": float(stock.get("change_percentage", "0%").replace("%", "")),
"volume": int(stock.get("volume", "0").replace(",", ""))
})
# Process top losers
losers = []
if "top_losers" in data:
for stock in data["top_losers"][:limit]:
losers.append({
"symbol": stock.get("ticker", ""),
"name": stock.get("name", ""),
"price": float(stock.get("price", "0").replace("$", "")),
"change": float(stock.get("change_amount", "0").replace("$", "")),
"change_percent": float(stock.get("change_percentage", "0%").replace("%", "")),
"volume": int(stock.get("volume", "0").replace(",", ""))
})
# Format the response
gainers_report = "Top Gainers:\n"
for idx, stock in enumerate(gainers, 1):
gainers_report += f"{idx}. {stock['symbol']} ({stock['name']}): +{stock['change_percent']}%, ${stock['price']}\n"
losers_report = "\nTop Losers:\n"
for idx, stock in enumerate(losers, 1):
losers_report += f"{idx}. {stock['symbol']} ({stock['name']}): {stock['change_percent']}%, ${stock['price']}\n"
full_report = gainers_report + losers_report
return {
"status": "success",
"report": full_report,
"gainers": gainers,
"losers": losers,
"last_updated": data.get("last_updated", "Unknown")
}
except Exception as e:
return {
"status": "error",
"error_message": f"Failed to retrieve market movers: {str(e)}"
}
def get_stock_details(symbol: str) -> Dict[str, Any]:
"""Retrieves detailed information for a specific stock.
Args:
symbol (str): The stock symbol to lookup (e.g., 'AAPL', 'MSFT')
Returns:
dict: Status and detailed information about the stock
"""
try:
# Get overview data
overview_url = f"https://www.alphavantage.co/query?function=OVERVIEW&symbol={symbol}&apikey={ALPHA_VANTAGE_API_KEY}"
overview_response = requests.get(overview_url)
if overview_response.status_code != 200:
return {
"status": "error",
"error_message": f"API request failed with status code {overview_response.status_code}"
}
overview_data = overview_response.json()
# Check if we got valid data
if "Symbol" not in overview_data:
return {
"status": "error",
"error_message": f"No data found for symbol {symbol}"
}
# Get quote data
quote_url = f"https://www.alphavantage.co/query?function=GLOBAL_QUOTE&symbol={symbol}&apikey={ALPHA_VANTAGE_API_KEY}"
quote_response = requests.get(quote_url)
if quote_response.status_code != 200:
return {
"status": "error",
"error_message": f"API request failed with status code {quote_response.status_code}"
}
quote_data = quote_response.json().get("Global Quote", {})
# Extract and format data
price = float(quote_data.get("05. price", "0"))
prev_close = float(quote_data.get("08. previous close", "0"))
change = float(quote_data.get("09. change", "0"))
change_percent = float(quote_data.get("10. change percent", "0%").replace("%", ""))
# Format the report
report = f"Stock Details for {symbol} ({overview_data.get('Name', symbol)}):\n"
report += f"Price: ${price}\n"
report += f"Change: {'+' if change >= 0 else ''}{change} ({'+' if change_percent >= 0 else ''}{change_percent}%)\n"
report += f"Sector: {overview_data.get('Sector', 'Unknown')}\n"
report += f"Industry: {overview_data.get('Industry', 'Unknown')}\n"
report += f"Market Cap: ${float(overview_data.get('MarketCapitalization', '0')) / 1e9:.2f} billion\n"
report += f"P/E Ratio: {overview_data.get('PERatio', 'N/A')}\n"
report += f"Dividend Yield: {overview_data.get('DividendYield', 'N/A')}\n"
report += f"52-week High: ${overview_data.get('52WeekHigh', 'N/A')}\n"
report += f"52-week Low: ${overview_data.get('52WeekLow', 'N/A')}\n"
report += f"Volume: {int(quote_data.get('06. volume', '0')):,}\n"
return {
"status": "success",
"report": report,
"symbol": symbol,
"name": overview_data.get('Name', symbol),
"price": price,
"change": change,
"change_percent": change_percent,
"sector": overview_data.get('Sector', 'Unknown'),
"industry": overview_data.get('Industry', 'Unknown'),
"description": overview_data.get('Description', '')
}
except Exception as e:
return {
"status": "error",
"error_message": f"Failed to retrieve stock details for {symbol}: {str(e)}"
}
# Create the agent
root_agent = Agent(
name="stock_market_agent",
model="gemini-2.0-flash",
description=(
"Agent to provide information about US stock market performance"
),
instruction=(
"You are a helpful financial agent who can answer user questions about the "
"stock market, retrieve information about top gainers and losers, and "
"provide details about specific stocks. with extra information and contex about the company"
"what you should look for in the stock market. weather you should invest in them or not including sources for extra information on the stock current CEO and latest information about the company and economic laws affect the stock ."
),
tools=[get_market_movers, get_stock_details],
)
Our root_agent uses the gemini-2.0-flash model to provide information to users based on their queries and uses the get_market_movers and get_stock_details as tools.
With these lines of code, our agent is ready to go live! We also have the extra option to communicate with our agent using the web interface simply by running the command below in the project root directory:
parent_folder/ <-- navigate to this directory
stock_agent/
__init__.py
agent.py
.env
If you are in the stock_agent directory, just run:
$ cd ..
$ adk web
You should see:
INFO: Started server process [94166]
INFO: Waiting for application startup.
+-----------------------------------------------------------------------------+
| ADK Web Server started |
| |
| For local testing, access at http://localhost:8000. |
+-----------------------------------------------------------------------------+
We can now interact with our agent via the UI at http://localhost:8000.
Select your agent, ask the agent for the top movers, and get additional information on a stock of interest.
You can also inspect function calls, responses, and model responses by clicking on the actions.
- Note: You can use the voice and video feature to interact with your agent as long as you are using an API model that allows it, such as the ‘gemini-2.0-flash-live-001’
And with that, our production-ready stock market agent is Live!!! The Google ADK creates user sessions; hence, we can have multiple users concurrently.
To deploy to Cloud Run, authenticate your gcloud CLI using the command below:
gcloud auth login
ADK packages your application so you can deploy it to Cloud Run using a simple command:
adk deploy cloud_run --with_ui stockAgent
You’ll see output similar to:
Start generating Cloud Run source files in /var/folders/11/b316r9b50230h38fqklj4rzr0000gn/T/cloud_run_deploy_src/20250413_193530
Copying agent source code...
Copying agent source code complete.
Creating Dockerfile...
Creating Dockerfile complete: /var/folders/11/b316r9b50230h38fqklj4rzr0000gn/T/cloud_run_deploy_src/20250413_193530/Dockerfile
Deploying to Cloud Run...
Select the region you want to deploy your app to
Deploying from source requires an Artifact Registry Docker repository to store built containers. A repository named [cloud-run-source-deploy] in region
[europe-west1] will be created.Do you want to continue (Y/n)? y
Allow unauthenticated invocations to [adk-default-service-name] (y/N)? y
Building using Dockerfile and deploying container to Cloud Run service [adk-default-service-name] in project [project-name] region [europe-west1]
⠧ Building and deploying new service... Uploading sources.
⠏ Building and deploying new service... Uploading sources.
⠧ Uploading sources...
✓ Building and deploying new service... Done.
. Creating Revision...
✓ Uploading sources...
✓ Building Container... Logs are available at [https://console.cloud.google.com/cloud-build/builds;region=europe-west1/977e-f9143e2
37e1c?].
✓ Creating Revision... -
✓ Routing traffic...
✓ Setting IAM Policy...
Done.
Service [adk-default-service-name] revision [adk-default-service-name-00001-jbr] has been deployed and is serving 100 percent of traffic.
Service URL: https://adk-default-service-name-90364875413.europe-west1.run.app
Cleaning up the temp folder: /var/folders/11/b316r9b50230h38fqklj4rzr0000gn/T/cloud_run_deploy_src/20250413_19353
You can check out my code here
References
Source Credit: https://medium.com/google-cloud/using-google-adk-to-build-and-deploy-a-production-ready-stock-market-ai-agent-to-cloud-run-in-0ffa69b51760?source=rss—-e52cf94d98af—4
