Autonomous Portfolio Analysis with Google ADK, Zerodha MCP, and LLMs

Modern financial analysis is rapidly moving toward automation and agentic workflows. Integrating large language models (LLMs) with real-time financial data unlocks not just powerful insights but also entirely new ways of interacting with portfolio data.

This post walks through a practical, autonomous solution using Google ADK, Zerodha’s Kite MCP protocol, and an LLM for actionable portfolio analytics. The full workflow and code are available on GitHub.

---

Why This Stack?

• Google ADK: Enables LLM agents to interact with live tools, APIs, and event streams in a repeatable, testable way.
• Zerodha MCP (Model Control Protocol): Provides a secure, real-time API to portfolio holdings using Server-Sent Events (SSE).
• LLMs (Gemini/GPT-4o): Analyze portfolio data, highlight concentration risk, and offer actionable recommendations.

---

Architecture Overview

The workflow has three main steps:

1. User authenticates with Zerodha using an OAuth browser flow.
2. The agent retrieves live holdings via the MCP get_holdings tool.
3. The LLM agent analyzes the raw data for risk and performance insights.

All API keys and connection details are managed through environment variables for security and reproducibility.

---

Key Code Snippets

1. Environment and Dependency Setup

import os
from dotenv import load_dotenv

# Load API keys and config from .env
load_dotenv('.env')
os.environ["GOOGLE_API_KEY"] = os.environ["GOOGLE_API_KEY"]
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "False"

---

2. ADK Agent and Toolset Initialization

from google.adk.agents.llm_agent import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, SseServerParams

MCP_SSE_URL = os.environ.get("MCP_SSE_URL", "https://mcp.kite.trade/sse")

toolset = MCPToolset(
    connection_params=SseServerParams(url=MCP_SSE_URL, headers={})
)

root_agent = LlmAgent(
    model='gemini-2.0-flash',
    name='zerodha_portfolio_assistant',
    instruction=(
        "You are an expert Zerodha portfolio assistant. "
        "Use the 'login' tool to authenticate, and the 'get_holdings' tool to fetch stock holdings. "
        "When given portfolio data, analyze for concentration risk and best/worst performers."
    ),
    tools=[toolset]
)

---

3. Orchestrating the Workflow

from google.adk.sessions import InMemorySessionService
from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService
from google.adk.runners import Runner
from google.genai import types

import asyncio

async def run_workflow():
    session_service = InMemorySessionService()
    artifacts_service = InMemoryArtifactService()
    session = await session_service.create_session(
        state={}, app_name='zerodha_portfolio_app', user_id='user1'
    )

    runner = Runner(
        app_name='zerodha_portfolio_app',
        agent=root_agent,
        artifact_service=artifacts_service,
        session_service=session_service,
    )

    # 1. Login Step
    login_query = "Authenticate and provide the login URL for Zerodha."
    content = types.Content(role='user', parts=[types.Part(text=login_query)])
    login_url = None
    async for event in runner.run_async(session_id=session.id, user_id=session.user_id, new_message=content):
        if event.is_final_response():
            import re
            match = re.search(r'(https?://[^\s)]+)', getattr(event.content.parts[0], "text", ""))
            if match:
                login_url = match.group(1)
    if not login_url:
        print("No login URL found. Exiting.")
        return
    print(f"Open this URL in your browser to authenticate:\n{login_url}")
    import webbrowser; webbrowser.open(login_url)
    input("Press Enter after completing login...")

    # 2. Fetch Holdings
    holdings_query = "Show my current stock holdings."
    content = types.Content(role='user', parts=[types.Part(text=holdings_query)])
    holdings_raw = None
    async for event in runner.run_async(session_id=session.id, user_id=session.user_id, new_message=content):
        if event.is_final_response():
            holdings_raw = getattr(event.content.parts[0], "text", None)
    if not holdings_raw:
        print("No holdings data found.")
        return

    # 3. Analysis
    analysis_prompt = f"""
You are a senior portfolio analyst.

Given only the raw stock holdings listed below, do not invent or assume any other holdings.

1. **Concentration Risk**: Identify if a significant percentage of the total portfolio is allocated to a single stock or sector. Quantify the largest exposures, explain why this matters, and suggest specific diversification improvements.

2. **Performance Standouts**: Clearly identify the best and worst performing stocks in the portfolio (by absolute and percentage P&L), and give actionable recommendations.

Raw holdings:

{holdings_raw}

Use only the provided data.
"""
    content = types.Content(role='user', parts=[types.Part(text=analysis_prompt)])
    async for event in runner.run_async(session_id=session.id, user_id=session.user_id, new_message=content):
        if event.is_final_response():
            print("\n=== Portfolio Analysis Report ===\n")
            print(getattr(event.content.parts[0], "text", ""))

asyncio.run(run_workflow())

---

Security and Environment Configuration

All API keys and MCP endpoints are managed via environment variables or a .env file.
Never hardcode sensitive information in code.

Example .env file:

GOOGLE_API_KEY=your_google_gemini_api_key
MCP_SSE_URL=https://mcp.kite.trade/sse

---

What This Enables

• Reproducible automation: Agents can authenticate, retrieve, and analyze portfolios with minimal human input.
• Extensibility: Easily add more tools (orders, margins, etc.) or more advanced analytic prompts.
• Separation of concerns: Business logic, security, and agent workflow are all clearly separated.

---

Repository

Full working code and documentation:
https://github.com/navveenb/agentic-ai-worfklows/tree/main/google-adk-zerodha

---

This workflow is for educational and portfolio analysis purposes only. Not investment advice.