Google ADK

import os
import asyncio
import scalekit.client
from dotenv import load_dotenv

load_dotenv()

scalekit_client = scalekit.client.ScalekitClient(
    client_id=os.getenv("SCALEKIT_CLIENT_ID"),
    client_secret=os.getenv("SCALEKIT_CLIENT_SECRET"),
    env_url=os.getenv("SCALEKIT_ENVIRONMENT_URL"),
)
actions = scalekit_client.actions

1
# Connect the user's Gmail account, and wait until it's ACTIVE before calling tools
2
connection_name = "gmail"
3
identifier = "user_123"  # your app's unique user ID
4

5
response = actions.get_or_create_connected_account(
6
    connection_name=connection_name, identifier=identifier
7
)
8
if response.connected_account.status != "ACTIVE":
9
    link = actions.get_authorization_link(
10
        connection_name=connection_name, identifier=identifier
11
    )
12
    print("Authorize Gmail:", link.link)
13
    input("Press Enter after authorizing...")
14
    # Fetch the account again to pick up the new status
15
    response = actions.get_or_create_connected_account(
16
        connection_name=connection_name, identifier=identifier
17
    )
18

19
if response.connected_account.status != "ACTIVE":
20
    raise RuntimeError(
21
        f"Gmail is {response.connected_account.status}, not ACTIVE. Authorize it and run again."
22
    )

1
from google.adk.agents import Agent
2
from google.adk.runners import Runner
3
from google.adk.sessions import InMemorySessionService
4
from google.genai import types
5

6
tools = actions.google.get_tools(
7
    identifier="user_123",
8
    connection_names=["gmail"],
9
    page_size=100,  # avoid missing tools when a connector has more than the default page
10
)
11

12
agent = Agent(
13
    name="gmail_assistant",
14
    model="gemini-2.0-flash",
15
    instruction="You are a helpful Gmail assistant.",
16
    tools=tools,
17
)
18

19
async def main():
20
    session_service = InMemorySessionService()
21
    runner = Runner(agent=agent, app_name="gmail_app", session_service=session_service)
22
    session = await session_service.create_session(app_name="gmail_app", user_id="user_123")
23

24
    message = types.Content(
25
        role="user",
26
        parts=[types.Part(text="Fetch my last 5 unread emails and summarize them")],
27
    )
28
    async for event in runner.run_async(
29
        user_id="user_123",
30
        session_id=session.id,
31
        new_message=message,
32
    ):
33
        if event.is_final_response() and event.content and event.content.parts:
34
            print(event.content.parts[0].text)
35

36
asyncio.run(main())

1
import os
2
from datetime import timedelta
3
from google.adk.agents import Agent
4
from google.adk.tools.mcp_tool import McpToolset, StreamableHTTPConnectionParams
5

6
# Returned by create_config when you created the Virtual MCP server
7
config_id = os.environ["SCALEKIT_MCP_CONFIG_ID"]
8
mcp_url = os.environ["SCALEKIT_MCP_SERVER_URL"]
9

10
# Mint a fresh session token for this user before each agent run
11
mcp_token = actions.mcp.create_session_token(
12
    mcp_config_id=config_id,
13
    identifier="user_123",
14
    expiry=timedelta(hours=1),
15
).token
16

17
agent = Agent(
18
    name="gmail_assistant",
19
    model="gemini-2.0-flash",
20
    instruction="You are a helpful Gmail assistant.",
21
    tools=[
22
        McpToolset(
23
            connection_params=StreamableHTTPConnectionParams(
24
                url=mcp_url,
25
                headers={"Authorization": f"Bearer {mcp_token}"},
26
            )
27
        )
28
    ],
29
)