Skip to main content

Email Agent

An example demonstrating how to build an email automation agent using the CAIP Agents SDK with context injection and Gmail integration. Check Full Code Here


Overview

This example shows:

  • ✅ Using context_schema for dependency injection
  • ✅ Tools with runtime context access (ToolRuntime)
  • ✅ External API integration (Gmail)
  • ✅ Both LangChain and PydanticAI patterns

Project Structure

email-agent/
├── app.py # Main entry point
├── email_agent.py # Agent configuration and tools
├── credentials.json # Gmail OAuth2 credentials
├── token.json # Gmail OAuth2 token
└── tools/
└── gmail_tools.py # Gmail API utilities

Setup

Install Dependencies

pip install caip-agents-sdk google-auth google-auth-oauthlib google-api-python-client

Gmail OAuth2 Setup

  1. Go to Google Cloud Console
  2. Create a new project or select existing
  3. Enable Gmail API (APIs & Services > Library)
  4. Create OAuth 2.0 Client ID (APIs & Services > Credentials > Desktop app)
  5. Download as credentials.json

Key Concepts

Context Schema (Dependency Injection)

The email agent uses context_schema to inject dependencies into tools at runtime:

from dataclasses import dataclass

@dataclass
class EmailAgentDependencies:
"""Dependencies for email agent execution."""
gmail_credentials_path: str
gmail_token_path: str
session_id: str = None

Pass the schema when creating the agent:

email_agent = client.create_agent(
framework="langchain",
agent_id="your-agent-id",
context_schema=EmailAgentDependencies, # Inject dependencies
)

Tools with Context

Tools access the injected context via ToolRuntime:

from caip_agents_sdk import ToolRuntime

@email_agent.tool
async def create_gmail_draft(
ctx: ToolRuntime[EmailAgentDependencies], # Context access
to: List[str],
subject: str,
body: str,
) -> Dict[str, Any]:
"""Creates a draft email in Gmail."""

# Access dependencies from context
credentials_path = ctx.context.gmail_credentials_path
token_path = ctx.context.gmail_token_path

# Use credentials...
service = await authenticate_gmail_service(credentials_path, token_path)
# Create draft...

Agent Configuration

# email_agent.py
from caip_agents_sdk import CAIPAgentsClient, ToolRuntime
from dataclasses import dataclass

@dataclass
class EmailAgentDependencies:
gmail_credentials_path: str
gmail_token_path: str
session_id: str = None

# Initialize client and agent
client = CAIPAgentsClient()

email_agent = client.create_agent(
framework="langchain",
agent_id="your-agent-id",
context_schema=EmailAgentDependencies,
)

# Register tools with context
@email_agent.tool
async def authenticate_gmail(ctx: ToolRuntime[EmailAgentDependencies]) -> dict:
"""Handles Gmail OAuth2 authentication."""
# Access ctx.context for credentials
pass

@email_agent.tool
async def create_gmail_draft(
ctx: ToolRuntime[EmailAgentDependencies],
to: list[str],
subject: str,
body: str,
) -> dict:
"""Creates a draft email in Gmail."""
pass

Running the Agent

# app.py
import asyncio
from email_agent import email_agent, EmailAgentDependencies

async def main():
# Create dependencies
deps = EmailAgentDependencies(
gmail_credentials_path="credentials.json",
gmail_token_path="token.json",
)

# Initialize agent
await email_agent.initialize()

# Run with context
result = await email_agent.run(
"Create a draft email for a leave request to manager@company.com",
context=deps, # Pass dependencies
)
print(result)

if __name__ == "__main__":
asyncio.run(main())

Run:

python app.py

Framework Comparison

LangChain (Default)

from caip_agents_sdk import ToolRuntime

email_agent = client.create_agent(
framework="langchain",
agent_id="...",
context_schema=EmailAgentDependencies,
)

@email_agent.tool
async def my_tool(ctx: ToolRuntime[EmailAgentDependencies], arg: str):
# Access: ctx.context.gmail_credentials_path
pass

# Run with context=
result = await email_agent.run("...", context=deps)

PydanticAI

from caip_agents_sdk import RunContext

email_agent = client.create_agent(
framework="pydantic_ai",
agent_id="...",
deps_type=EmailAgentDependencies, # Note: deps_type instead of context_schema
)

@email_agent.tool
async def my_tool(ctx: RunContext[EmailAgentDependencies], arg: str):
# Access: ctx.deps.gmail_credentials_path
pass

# Run with deps=
result = await email_agent.run("...", deps=deps)

Next Steps