LangChain

Build a LangChain agent that reads a user’s Gmail inbox. Scalekit handles OAuth, token storage, and returns tools in native LangChain format. Your agent code needs no Scalekit-specific logic beyond initialization.

Full code on GitHub

Install

pip install scalekit-sdk-python langchain-openai

Initialize

import os
import scalekit.client

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

Connect the user to Gmail

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

response = actions.get_or_create_connected_account(
    connection_name=connection_name, identifier=identifier
)
if response.connected_account.status != "ACTIVE":
    link = actions.get_authorization_link(
        connection_name=connection_name, identifier=identifier
    )
    print("Authorize Gmail:", link.link)
    input("Press Enter after authorizing...")
    # Fetch the account again to pick up the new status
    response = actions.get_or_create_connected_account(
        connection_name=connection_name, identifier=identifier
    )

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

See Authorize a user for production auth handling.

Build and run the agent

actions.langchain.get_tools() returns native StructuredTool objects. Bind them to your LLM and run the tool-calling loop:

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, ToolMessage

tools = actions.langchain.get_tools(
    identifier="user_123",
    connection_names=["gmail"],
    page_size=100,  # avoid missing tools when a connector has more than the default page
)
tool_map = {t.name: t for t in tools}

llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)
messages = [HumanMessage("Fetch my last 5 unread emails and summarize them")]

while True:
    response = llm.invoke(messages)
    messages.append(response)
    if not response.tool_calls:
        print(response.content)
        break
    for tc in response.tool_calls:
        result = tool_map[tc["name"]].invoke(tc["args"])
        messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))

Multiple Gmail accounts

If a user has multiple Gmail connections, pass the specific connection_names value from your Scalekit dashboard to scope tools to the right one.

Use MCP instead

LangChain connects to MCP servers with langchain-mcp-adapters. Pass the Virtual MCP Server URL and a session token for this user:

pip install "langchain-mcp-adapters>=0.3,<1"

import asyncio
import os
from datetime import timedelta
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, ToolMessage

# Returned by create_config when you created the Virtual MCP Server (see Set up and connect)
config_id = os.environ["SCALEKIT_MCP_CONFIG_ID"]
mcp_url = os.environ["SCALEKIT_MCP_SERVER_URL"]

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

async def run():
    client = MultiServerMCPClient(
        {
            "scalekit": {
                "transport": "streamable_http",
                "url": mcp_url,
                "headers": {"Authorization": f"Bearer {mcp_token}"},
            }
        }
    )
    tools = await client.get_tools()
    tool_map = {t.name: t for t in tools}
    llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)
    messages = [HumanMessage("Fetch my last 5 unread emails and summarize them")]

    while True:
        response = await llm.ainvoke(messages)
        messages.append(response)
        if not response.tool_calls:
            print(response.content)
            break
        for tc in response.tool_calls:
            result = await tool_map[tc["name"]].ainvoke(tc["args"])
            messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))

asyncio.run(run())

See Set up and connect to create the server and check that the user’s connections are active before you mint a token.