Gmail Labeler

1
from composio import Composio
2
from composio_langchain import LangchainProvider
3
4
from langchain import hub
5
from langchain.agents import AgentExecutor, create_openai_functions_agent
6
from langchain_openai import ChatOpenAI
7
8
9
def create_agent(user_id: str, composio_client: Composio[LangchainProvider]):
10
    """
11
    Create an agent for a given user id.
12
    """
13
    # Step 1: Get all the tools
14
    tools = composio_client.tools.get(
15
        user_id=user_id,
16
        tools=[
17
            "GMAIL_LIST_LABELS",
18
            "GMAIL_ADD_LABEL_TO_EMAIL",
19
            "GMAIL_CREATE_LABEL",
20
        ],
21
    )
22
23
    # Step 2: Pull relevant agent prompt.
24
    prompt = hub.pull("hwchase17/openai-functions-agent")
25
26
    # Step 3: Initialize chat model.
27
    openai_client = ChatOpenAI(model="gpt-4-turbo")
28
29
    # Step 4: Define agent
30
    return AgentExecutor(
31
        agent=create_openai_functions_agent(
32
            openai_client,
33
            tools,
34
            prompt,
35
        ),
36
        tools=tools,
37
        verbose=False,
38
    )

1
# Function to initiate a connected account
2
def create_connection(composio_client: Composio[OpenAIProvider], user_id: str):
3
    """
4
    Create a connection for a given user id and auth config id.
5
    """
6
    # Fetch or create the auth config for the gmail toolkit
7
    auth_config = fetch_auth_config(composio_client=composio_client)
8
    if not auth_config:
9
        auth_config = create_auth_config(composio_client=composio_client)
10
11
    # Create a connection for the user
12
    return composio_client.connected_accounts.initiate(
13
        user_id=user_id,
14
        auth_config_id=auth_config.id,
15
    )

1
def check_connected_account_exists(
2
    composio_client: Composio[LangchainProvider],
3
    user_id: str,
4
):
5
    """
6
    Check if a connected account exists for a given user id.
7
    """
8
    # Fetch all connected accounts for the user
9
    connected_accounts = composio_client.connected_accounts.list(
10
        user_ids=[user_id],
11
        toolkit_slugs=["GMAIL"],
12
    )
13
14
    # Check if there's an active connected account
15
    for account in connected_accounts.items:
16
        if account.status == "ACTIVE":
17
            return True
18
19
        # Ideally you should not have inactive accounts, but if you do, you should delete them
20
        print(f"[warning] inactive account {account.id} found for user id: {user_id}")
21
    return False

1
# Create a new trigger
2
def create_trigger(
3
    composio_client: Composio[LangchainProvider],
4
    connected_account_id: str,
5
) -> str:
6
    """
7
    Create a trigger.
8
    """
9
    response = composio_client.triggers.create(
10
        slug="GMAIL_NEW_GMAIL_MESSAGE",
11
        connected_account_id=connected_account_id,
12
        trigger_config={},
13
    )
14
    return response.trigger_id

1
def check_trigger_exists(
2
    composio_client: Composio[LangchainProvider],
3
    connected_account_id: str,
4
) -> t.Optional[str]:
5
    """
6
    Check if a trigger exists.
7
    """
8
    triggers = composio_client.triggers.list_active(
9
        trigger_names=["GMAIL_NEW_GMAIL_MESSAGE"],
10
        connected_account_ids=[connected_account_id],
11
    )
12
    for trigger in triggers.items:
13
        return trigger.id
14
    return None

1
# Create subscription object
2
trigger_subscription = composio_client.triggers.subscribe()
3
4
# Register event handler
5
@trigger_subscription.handle(
6
    trigger_id="<TRIGGER_ID>",  # Filter out events that does not belong this trigger id
7
    trigger_slug="GMAIL_NEW_GMAIL_MESSAGE",
8
)
9
def handle_event(event: TriggerEvent):
10
    print("> Received email with subject: ", event["payload"]["subject"])
11
12
# Wait for events
13
trigger_subscription.wait_forever()

1
# Create a trigger subscription factory
2
3
def create_trigger_subscription(
4
    composio_client: Composio[LangchainProvider],
5
    trigger_slug: str,
6
    trigger_id: str,
7
    agent: AgentExecutor,
8
):
9
    """
10
    Create a trigger subscription for the given agent.
11
    """
12
    trigger_subscription = composio_client.triggers.subscribe()
13
14
    @trigger_subscription.handle(
15
        trigger_slug=trigger_slug,
16
        trigger_id=trigger_id,
17
    )
18
    def handle_event(event: TriggerEvent):
19
        print("> Received email with subject: ", event["payload"]["subject"])
20
        result = agent.invoke(
21
            input={
22
                "input": APPLY_NEW_LABEL.format(  # Check `gmail_labeler/prompt.py`
23
                    message_id=event["payload"]["id"],
24
                    message_subject=event["payload"]["subject"],
25
                    message_text=event["payload"]["message_text"],
26
                )
27
            }
28
        )
29
        print("> Result: ", result["output"])
30
31
    return trigger_subscription

1
def run_agent(user_id: str):
2
    # Create composio client
3
    composio_client = Composio(provider=LangchainProvider())
4
5
    # Validate conected account
6
    connected_account_id = check_connected_account_exists(composio_client, user_id)
7
    if connected_account_id is None:
8
        connection_request = create_connection(composio_client, user_id)
9
        print(
10
            f"Authenticate with the following link: {connection_request.redirect_url}"
11
        )
12
        connection_request.wait_for_connection()
13
        connected_account_id = connection_request.id
14
15
    # Check if trigger exists, create if not
16
    trigger_id = check_trigger_exists(
17
        composio_client=composio_client,
18
        connected_account_id=connected_account_id,
19
    )
20
    if trigger_id is None:
21
        trigger_id = create_trigger(
22
            composio_client=composio_client,
23
            connected_account_id=connected_account_id,
24
        )
25
26
    # Create agent
27
    agent = create_agent(user_id=user_id, composio_client=composio_client)
28
29
    # Create trigger subscription
30
    trigger_subscription = create_trigger_subscription(
31
        composio_client=composio_client,
32
        trigger_slug=GMAIL_NEW_GMAIL_MESSAGE_TRIGGER,
33
        trigger_id=trigger_id,
34
        agent=agent,
35
    )
36
37
    # Wait forever
38
    print("Waiting for events...")
39
    trigger_subscription.wait_forever()