Google ADK

pip install google-adk

uv add google-adk

import os
import base64
import asyncio
from google.adk import Runner
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import McpToolset, StreamableHTTPConnectionParams
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Configuration - API keys from environment (secrets)
STACKONE_API_KEY = os.getenv("STACKONE_API_KEY")
os.environ["GOOGLE_API_KEY"] = os.getenv("GOOGLE_API_KEY")

# Encode API key for Basic auth
auth_token = base64.b64encode(f"{STACKONE_API_KEY}:".encode()).decode()

def create_stackone_agent(account_id: str) -> LlmAgent:
    """
    Create an agent for a specific StackOne account.

    In production, account_id should come from:
    - User/tenant context
    - Database lookup based on authenticated user
    - Request parameters
    """
    return LlmAgent(
        model="gemini-2.5-pro",
        name="stackone_assistant",
        instruction="You are a helpful assistant with access to data from connected platforms via StackOne.",
        tools=[
            McpToolset(
                connection_params=StreamableHTTPConnectionParams(
                    url="https://api.stackone.com/mcp",
                    headers={
                        "Authorization": f"Basic {auth_token}",
                        "x-account-id": account_id
                    }
                ),
                tool_name_prefix="stackone_"
            )
        ],
    )

async def main():
    # Get account_id from user context (not environment variable)
    account_id = get_user_account_id()  # Your function to determine this

    # Create agent for this specific account
    agent = create_stackone_agent(account_id)

    # Create runner with in-memory session service
    runner = Runner(
        agent=agent,
        app_name="stackone_demo",
        session_service=InMemorySessionService()
    )

    # Create a session
    await runner.session_service.create_session(
        app_name="stackone_demo",
        user_id="user_1",
        session_id="session_1"
    )

    # Create and send a message
    message = types.Content(parts=[types.Part(text="List workers from Workday, limit 5")], role="user")

    # Run the agent and collect response
    async for event in runner.run_async(
        user_id="user_1",
        session_id="session_1",
        new_message=message
    ):
        if hasattr(event, 'content') and event.content:
            if hasattr(event.content, 'parts'):
                for part in event.content.parts:
                    if hasattr(part, 'text') and part.text:
                        print(part.text)

asyncio.run(main())

export STACKONE_API_KEY="your_stackone_api_key"
export GOOGLE_API_KEY="your_google_api_key"

# First query
message1 = types.Content(parts=[types.Part(text="List 3 workers")], role="user")
async for event in runner.run_async(user_id="user_1", session_id="session_1", new_message=message1):
    # Process first response...
    pass

# Follow-up query (uses same session for context)
message2 = types.Content(parts=[types.Part(text="Show me details for the first worker")], role="user")
async for event in runner.run_async(user_id="user_1", session_id="session_1", new_message=message2):
    # Process follow-up response...
    pass

McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://api.stackone.com/mcp",
        headers={
            "Authorization": f"Basic {auth_token}",
            "x-account-id": account_id
        }
    ),
    tool_filter=[
        "workday_list_workers",
        "workday_get_worker",
        "salesforce_crm_list_accounts"
    ]  # Only expose specific tools
)