Arcade Client Implementation: Sync and Async (#22)

This PR introduces both synchronous and asynchronous Arcade client
implementations, providing a robust interface for interacting with the
Arcade API.

## Key Features

1. Synchronous (`SyncArcade`) and Asynchronous (`AsyncArcade`) clients
2. Authentication and Tool resources
3. OpenAI chat completions integration
4. Comprehensive error handling

## Client Methods

Both `SyncArcade` and `AsyncArcade` offer:

- `auth.authorize()`: Initiate authorization
- `auth.poll_authorization()`: Check authorization status
- `tool.run()`: Execute a tool
- `tool.get()`: Retrieve tool specification
- `chat.create()`: Create chat completions

## Usage Examples

### Synchronous Authorization

```python
from arcade.client import AuthProvider, SyncArcade

client = SyncArcade(base_url="https://api.arcade.com", api_key="your_api_key")
auth_response = client.auth.authorize(
    provider=AuthProvider.google,
    scopes=["https://www.googleapis.com/auth/gmail.readonly"],
    user_id="user123"
)
print(f"Authorize at: {auth_response.auth_url}")
```

### Asynchronous Authorization

```python
import asyncio
from arcade.client import AuthProvider, AsyncArcade

async def authorize():
    client = AsyncArcade(base_url="https://api.arcade.com", api_key="your_api_key")
    auth_response = await client.auth.authorize(
        provider=AuthProvider.slack_user,
        scopes=["chat:write", "im:write"],
        user_id="user456"
    )
    print(f"Authorize at: {auth_response.auth_url}")

asyncio.run(authorize())
```

This implementation provides a flexible and powerful way to interact
with Arcade services, supporting both synchronous and asynchronous
workflows.

arcade/arcade/actor/core/base.py

@@ -82,7 +82,7 @@ class BaseActor(Actor):
         duration_ms = (end_time - start_time) * 1000  # Convert to milliseconds
 
         return ToolCallResponse(
-            invocation_id=tool_request.invocation_id,
+            invocation_id=tool_request.invocation_id or "",
             duration=duration_ms,
             finished_at=datetime.now().isoformat(),
             success=not output.error,

arcade/arcade/actor/core/common.py

@@ -42,14 +42,23 @@ class Actor(ABC):
 
     @abstractmethod
     def get_catalog(self) -> list[ToolDefinition]:
+        """
+        Get the catalog of tools available in the actor.
+        """
         pass
 
     @abstractmethod
     async def call_tool(self, request: ToolCallRequest) -> ToolCallResponse:
+        """
+        Send a request to call a tool to the Actor
+        """
         pass
 
     @abstractmethod
     def health_check(self) -> dict[str, Any]:
+        """
+        Perform a health check of the actor
+        """
         pass
 
 

arcade/arcade/actor/fastapi/actor.py

@@ -49,7 +49,6 @@ class FastAPIRouter(Router):
                 method=request.method,
                 body_json=body_json,
             )
-
             if is_async_callable(handler):
                 return await handler(request_data)
             else:

arcade/arcade/client/__init__.py

@@ -0,0 +1,8 @@
+from arcade.client.client import Arcade, AsyncArcade
+from arcade.client.schema import AuthProvider
+
+__all__ = [
+    "AuthProvider",
+    "AsyncArcade",
+    "Arcade",
+]

arcade/arcade/client/base.py

@@ -0,0 +1,150 @@
+import os
+from typing import Any, Generic, TypeVar
+from urllib.parse import urljoin
+
+import httpx
+from httpx import Timeout
+
+from arcade.core.config import config
+
+T = TypeVar("T")
+ResponseT = TypeVar("ResponseT")
+
+
+class BaseResource(Generic[T]):
+    """Base class for all resources."""
+
+    def __init__(self, client: T):
+        self._client = client
+
+
+class BaseArcadeClient:
+    """Base class for Arcade clients."""
+
+    def __init__(
+        self,
+        base_url: str,
+        api_key: str | None = None,
+        headers: dict[str, str] | None = None,
+        proxies: str | dict[str, str] | None = None,
+        timeout: float | Timeout = 10.0,
+        retries: int = 3,
+    ):
+        """
+        Initialize the BaseArcadeClient.
+
+        Args:
+            base_url: The base URL for the Arcade API.
+            api_key: The API key for authentication.
+            headers: Additional headers to include in requests.
+            proxies: Proxy configuration for requests.
+            timeout: Request timeout in seconds.
+            retries: Number of retries for failed requests.
+        """
+        self._base_url = base_url
+        self._api_key = api_key or os.environ.get("ARCADE_API_KEY") or config.api.key
+        self._headers = headers or {}
+        self._headers.setdefault("X-API-Key", self._api_key)
+        self._headers.setdefault("Content-Type", "application/json")
+        self._proxies = proxies
+        self._timeout = timeout
+        self._retries = retries
+
+    def _build_url(self, path: str) -> str:
+        """
+        Build the full URL for a given path.
+
+        Args:
+            path: The path to append to the base URL.
+
+        Returns:
+            The full URL.
+        """
+        return urljoin(self._base_url, path)
+
+
+class SyncArcadeClient(BaseArcadeClient):
+    """Synchronous Arcade client."""
+
+    def __init__(self, *args: Any, **kwargs: Any):
+        super().__init__(*args, **kwargs)
+        self._client = httpx.Client(
+            base_url=self._base_url,
+            headers=self._headers,
+            proxies=self._proxies,
+            timeout=self._timeout,
+        )
+
+    def _request(self, method: str, path: str, **kwargs: Any) -> httpx.Response:
+        """
+        Make a synchronous HTTP request.
+        """
+        url = self._build_url(path)
+        for attempt in range(self._retries):
+            try:
+                print(method, url, kwargs)
+                response = self._client.request(method, url, **kwargs)
+                response.raise_for_status()
+                return response  # noqa: TRY300
+            except httpx.HTTPStatusError:
+                if attempt == self._retries - 1:
+                    raise
+        raise RuntimeError("This should never be reached")
+
+    def close(self) -> None:
+        """Close the client session."""
+        self._client.close()
+
+    def __enter__(self) -> "SyncArcadeClient":
+        return self
+
+    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+        self.close()
+
+
+class AsyncArcadeClient(BaseArcadeClient):
+    """Asynchronous Arcade client."""
+
+    def __init__(self, *args: Any, **kwargs: Any):
+        super().__init__(*args, **kwargs)
+        self._client: httpx.AsyncClient | None = None
+
+    async def _get_client(self) -> httpx.AsyncClient:
+        """
+        Get or create an asynchronous HTTP client.
+        """
+        if self._client is None:
+            self._client = httpx.AsyncClient(
+                base_url=self._base_url,
+                headers=self._headers,
+                proxies=self._proxies,
+                timeout=self._timeout,
+            )
+        return self._client
+
+    async def _request(self, method: str, path: str, **kwargs: Any) -> httpx.Response:
+        """
+        Make an asynchronous HTTP request.
+        """
+        client = await self._get_client()
+        url = self._build_url(path)
+        for attempt in range(self._retries):
+            try:
+                response = await client.request(method, url, **kwargs)
+                response.raise_for_status()
+                return response  # noqa: TRY300
+            except httpx.HTTPStatusError:
+                if attempt == self._retries - 1:
+                    raise
+        raise RuntimeError("This should never be reached")
+
+    async def close(self) -> None:
+        """Close the client session."""
+        if self._client:
+            await self._client.aclose()
+
+    async def __aenter__(self) -> "AsyncArcadeClient":
+        return self
+
+    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+        await self.close()

arcade/arcade/client/client.py

@@ -0,0 +1,215 @@
+from typing import Any, Generic, TypeVar
+
+import httpx
+from openai import AsyncOpenAI, OpenAI
+from openai.resources.chat import AsyncChat, Chat
+
+from arcade.client.base import AsyncArcadeClient, BaseResource, SyncArcadeClient
+from arcade.client.errors import (
+    APIStatusError,
+    BadRequestError,
+    InternalServerError,
+    NotFoundError,
+    PermissionDeniedError,
+    UnauthorizedError,
+)
+from arcade.client.schema import (
+    AuthProvider,
+    AuthRequest,
+    AuthResponse,
+    ExecuteToolResponse,
+)
+from arcade.core.schema import ToolDefinition
+
+T = TypeVar("T")
+ClientT = TypeVar("ClientT", SyncArcadeClient, AsyncArcadeClient)
+
+
+class AuthResource(BaseResource[ClientT]):
+    """Authentication resource."""
+
+    _base_path = "/v1/auth"
+
+    def authorize(
+        self,
+        provider: AuthProvider,
+        scopes: list[str],
+        user_id: str,
+        authority: str | None = None,
+    ) -> AuthResponse:
+        """
+        Initiate an authorization request.
+
+        Args:
+            provider: The authorization provider.
+            scopes: The scopes required for the authorization.
+            user_id: The user ID initiating the authorization.
+            authority: The authority initiating the authorization.
+        """
+        auth_provider = provider.value
+
+        body = {
+            "auth_requirement": {
+                "provider": auth_provider,
+                auth_provider: AuthRequest(scope=scopes, authority=authority).dict(
+                    exclude_none=True
+                ),
+            },
+            "user_id": user_id,
+        }
+
+        data = self._client._execute_request(  # type: ignore[attr-defined]
+            "POST",
+            f"{self._base_path}/authorize",
+            json=body,
+        )
+        return AuthResponse(**data)
+
+    def poll_authorization(self, auth_id: str) -> AuthResponse:
+        """
+        Poll for the status of an authorization request.
+
+        Args:
+            auth_id: The authorization ID.
+
+        Example:
+            auth_status = client.auth.poll_authorization("auth_123")
+        """
+        data = self._client._execute_request(  # type: ignore[attr-defined]
+            "GET", f"{self._base_path}/status", params={"authorizationID": auth_id}
+        )
+        return AuthResponse(**data)
+
+
+class ToolResource(BaseResource[ClientT]):
+    """Tool resource."""
+
+    _base_path = "/v1/tools"
+
+    def run(
+        self,
+        tool_name: str,
+        user_id: str,
+        tool_version: str | None = None,
+        inputs: dict[str, Any] | None = None,
+    ) -> ExecuteToolResponse:
+        """
+        Send a request to execute a tool and return the response.
+
+        Args:
+            tool_name: The name of the tool to execute.
+            user_id: The user ID initiating the tool execution.
+            tool_version: The version of the tool to execute (if not provided, the latest version will be used).
+            inputs: The inputs for the tool.
+        """
+        request_data = {
+            "tool_name": tool_name,
+            "user_id": user_id,
+            "tool_version": tool_version,
+            "inputs": inputs,
+        }
+        data = self._client._execute_request(  # type: ignore[attr-defined]
+            "POST", f"{self._base_path}/execute", json=request_data
+        )
+        return ExecuteToolResponse(**data)
+
+    def get(self, director_id: str, tool_id: str) -> ToolDefinition:
+        """
+        Get the specification for a tool.
+
+        Args:
+            director_id: The director ID.
+            tool_id: The tool ID.
+        """
+        data = self._client._execute_request(  # type: ignore[attr-defined]
+            "GET",
+            f"{self._base_path}/definition",
+            params={"director_id": director_id, "tool_id": tool_id},
+        )
+        return ToolDefinition(**data)
+
+
+class ArcadeClientMixin(Generic[ClientT]):
+    """Mixin for Arcade clients."""
+
+    def __init__(self, base_url: str, *args: Any, **kwargs: Any):
+        super().__init__(base_url, *args, **kwargs)
+        self._openai_client: OpenAI | AsyncOpenAI | None = None
+        self.auth: AuthResource = AuthResource(self)
+        self.tool: ToolResource = ToolResource(self)
+        self.chat: Chat | AsyncChat | None = None
+
+    def _handle_http_error(
+        self,
+        e: httpx.HTTPStatusError,
+        error_map: dict[int, type[APIStatusError]],
+    ) -> None:
+        status_code = e.response.status_code
+        error_class = error_map.get(status_code, InternalServerError)
+        raise error_class(str(e), response=e.response)
+
+
+class Arcade(ArcadeClientMixin[SyncArcadeClient], SyncArcadeClient):
+    """Synchronous Arcade client."""
+
+    def __init__(self, *args: Any, **kwargs: Any):
+        super().__init__(*args, **kwargs)
+        self._openai_client = OpenAI(base_url=self._base_url + "/v1")
+        self.chat = self._openai_client.chat
+
+    def _execute_request(self, method: str, url: str, **kwargs: Any) -> Any:
+        """
+        Execute a synchronous request.
+
+        Args:
+            method: The HTTP method.
+            url: The URL to request.
+            **kwargs: Additional arguments for the request.
+        """
+        try:
+            response = self._request(method, url, **kwargs)
+            return response.json()
+        except httpx.HTTPStatusError as e:
+            self._handle_http_error(
+                e,
+                {
+                    400: BadRequestError,
+                    401: UnauthorizedError,
+                    403: PermissionDeniedError,
+                    404: NotFoundError,
+                    500: InternalServerError,
+                },
+            )
+
+
+class AsyncArcade(ArcadeClientMixin[AsyncArcadeClient], AsyncArcadeClient):
+    """Asynchronous Arcade client."""
+
+    def __init__(self, *args: Any, **kwargs: Any):
+        super().__init__(*args, **kwargs)
+        self._openai_client = AsyncOpenAI(base_url=self._base_url + "/v1")
+        self.chat = self._openai_client.chat
+
+    async def _execute_request(self, method: str, url: str, **kwargs: Any) -> Any:
+        """
+        Execute an asynchronous request.
+
+        Args:
+            method: The HTTP method.
+            url: The URL to request.
+            **kwargs: Additional arguments for the request.
+        """
+        try:
+            response = await self._request(method, url, **kwargs)
+            return response.json()
+        except httpx.HTTPStatusError as e:
+            self._handle_http_error(
+                e,
+                {
+                    400: BadRequestError,
+                    401: UnauthorizedError,
+                    403: PermissionDeniedError,
+                    404: NotFoundError,
+                    500: InternalServerError,
+                },
+            )

arcade/arcade/client/errors.py

@@ -0,0 +1,76 @@
+from typing import Optional
+
+import httpx
+
+
+class ArcadeError(Exception):
+    """Top-level exception for Arcade Client errors."""
+
+    pass
+
+
+class APIError(ArcadeError):
+    """Base class for API-related errors."""
+
+    def __init__(self, message: str, request: httpx.Request, *, body: Optional[object] = None):
+        super().__init__(message)
+        self.message = message
+        self.request = request
+        self.body = body
+
+
+class APIStatusError(APIError):
+    """Raised when an API response has a status code of 4xx or 5xx."""
+
+    def __init__(self, message: str, *, response: httpx.Response, body: Optional[object] = None):
+        super().__init__(message, response.request, body=body)
+        self.response = response
+        self.status_code = response.status_code
+
+
+class BadRequestError(APIStatusError):
+    """400 Bad Request"""
+
+    status_code = 400
+
+
+class UnauthorizedError(APIStatusError):
+    """401 Unauthorized"""
+
+    status_code = 401
+
+
+class PermissionDeniedError(APIStatusError):
+    """403 Forbidden"""
+
+    status_code = 403
+
+
+class NotFoundError(APIStatusError):
+    """404 Not Found"""
+
+    status_code = 404
+
+
+class RateLimitError(APIStatusError):
+    """429 Too Many Requests"""
+
+    status_code = 429
+
+
+class InternalServerError(APIStatusError):
+    """500 Internal Server Error"""
+
+    status_code = 500
+
+
+class UnprocessableEntityError(APIStatusError):
+    """422 Unprocessable Entity"""
+
+    status_code = 422
+
+
+class ServiceUnavailableError(APIStatusError):
+    """503 Service Unavailable"""
+
+    status_code = 503

arcade/arcade/client/schema.py

@@ -0,0 +1,76 @@
+from enum import Enum
+
+from pydantic import AnyUrl, BaseModel, Field
+
+from arcade.core.schema import ToolCallOutput, ToolContext
+
+
+class AuthProvider(str, Enum):
+    """The supported authorization providers."""
+
+    oauth2 = "oauth2"
+    """OAuth 2.0 authorization"""
+
+    google = "google"
+    """Google authorization"""
+
+    slack_user = "slack_user"
+    """Slack (user token) authorization"""
+
+    github_app = "github_app"
+    """GitHub App authorization"""
+
+
+class AuthRequest(BaseModel):
+    """
+    The requirements for authorization for a tool
+    """
+
+    authority: AnyUrl | None = None
+    """The URL of the OAuth 2.0 authorization server."""
+
+    scope: list[str]
+    """The scope(s) needed for authorization."""
+
+
+class AuthStatus(str, Enum):
+    """The status of an authorization request."""
+
+    pending = "pending"
+    failed = "failed"
+    completed = "completed"
+
+
+class AuthResponse(BaseModel):
+    """Response from an authorization request."""
+
+    auth_id: str = Field(alias="authorizationID")
+    """The ID of the authorization request"""
+
+    # TODO: Use AnyUrl?
+    auth_url: str | None = Field(None, alias="authorizationURL")
+    """The URL for the authorization"""
+
+    status: AuthStatus
+    """Only completed implies presence of a token"""
+
+    context: ToolContext | None = None
+
+
+class ExecuteToolResponse(BaseModel):
+    """Response from executing a tool."""
+
+    invocation_id: str
+    """The globally-unique ID for this tool invocation in the run."""
+
+    duration: float
+    """The duration of the tool invocation in milliseconds."""
+
+    finished_at: str
+    """The timestamp when the tool invocation finished."""
+
+    success: bool
+    """Whether the tool invocation was successful."""
+
+    output: ToolCallOutput | None = None
+    """The output of the tool invocation."""

arcade/arcade/core/catalog.py

@@ -86,6 +86,10 @@ class MaterializedTool(BaseModel):
     def description(self) -> str:
         return self.definition.description
 
+    @property
+    def requires_auth(self) -> bool:
+        return self.definition.requirements.authorization is not None
+
 
 class ToolCatalog(BaseModel):
     """Singleton class that holds all tools for a given actor"""

arcade/arcade/core/schema.py

@@ -147,11 +147,11 @@ class ToolContext(BaseModel):
 class ToolCallRequest(BaseModel):
     """The request to call (invoke) a tool."""
 
-    run_id: str
+    run_id: str | None = None
     """The globally-unique run ID provided by the Engine."""
-    invocation_id: str
+    invocation_id: str | None = None
     """The globally-unique ID for this tool invocation in the run."""
-    created_at: str
+    created_at: str | None = None
     """The timestamp when the tool invocation was created."""
     tool: ToolVersion
     """The name and version of the tool."""

examples/langchain/gmail.py

@@ -0,0 +1,65 @@
+import os
+
+from google.oauth2.credentials import Credentials
+from langchain_google_community import GmailToolkit
+from langchain_google_community.gmail.utils import (
+    build_resource_service,
+)
+from langchain_openai import ChatOpenAI
+from langgraph.prebuilt import create_react_agent
+
+# Step 1: Install required packages
+# Run the following in your terminal:
+# %pip install -qU langchain-google-community[gmail]
+# %pip install -qU langchain-openai
+# %pip install -qU langgraph
+#
+# Step 2: Set environment variables for LangChain and OpenAI API keys
+# Uncomment the following lines if you have the LangSmith API key
+# os.environ["LANGCHAIN_TRACING_V2"] = "true"
+# os.environ["LANGCHAIN_API_KEY"] = getpass.getpass("Enter your LangSmith API key: ")
+# Step 3: Authenticate with Gmail
+# credentials = get_gmail_credentials(
+#    token_file="token.json",
+#    scopes=["https://mail.google.com/"],
+#    client_secrets_file="credentials.json",
+# )
+# alternative way to authenticate with arcade
+from arcade.client import Arcade, AuthProvider
+
+client = Arcade(base_url="http://localhost:9099", api_key=os.environ["ARCADE_API_KEY"])
+
+challenge = client.auth.authorize(
+    provider=AuthProvider.google,
+    scopes=["https://www.googleapis.com/auth/gmail.readonly"],
+    user_id="example_user_id",
+)
+
+if challenge.status != "completed":
+    print(f"Please visit this URL to authorize: {challenge.auth_url}")
+    input("Press Enter after you've completed the authorization...")
+    challenge = client.auth.poll_authorization(challenge.auth_id)
+    if challenge.status != "completed":
+        print("Authorization not completed. Please try again.")
+        exit(1)
+
+
+creds = Credentials(challenge.context.authorization.token)
+api_resource = build_resource_service(credentials=creds)
+toolkit = GmailToolkit(api_resource=api_resource)
+
+# Step 4: Get available tools
+tools = toolkit.get_tools()
+
+# Step 5: Initialize the LLM and create an agent
+llm = ChatOpenAI(model="gpt-4o")
+agent_executor = create_react_agent(llm, tools)
+
+# Step 6: Draft an email using the agent
+example_query = "Read my latest emails to me and summarize them."
+events = agent_executor.stream(
+    {"messages": [("user", example_query)]},
+    stream_mode="values",
+)
+for event in events:
+    event["messages"][-1].pretty_print()

toolkits/gmail/arcade_gmail/tools/gmail.py

@@ -1,8 +1,9 @@
 import base64
-from email.mime.text import MIMEText
+import json
 import re
 from base64 import urlsafe_b64decode
-from typing import Annotated
+from email.mime.text import MIMEText
+from typing import Annotated, Any, Dict, Optional
 
 from bs4 import BeautifulSoup
 from google.oauth2.credentials import Credentials
@@ -55,75 +56,134 @@ def get_draft_url(draft_id):
 async def get_emails(
     context: ToolContext,
     n_emails: Annotated[int, "Number of emails to read"] = 5,
-) -> dict[str, list[dict[str, str]]]:
+) -> Annotated[str, "A list of email details in JSON format"]:
-    """Read emails from a Gmail account and extract plain text content, removing any HTML."""
+    """
+    Read emails from a Gmail account and extract plain text content.
 
-    # Build the Gmail service using the provided OAuth2 token
+    Args:
+        context (ToolContext): The context containing authorization information.
+        n_emails (int): Number of emails to read (default: 5).
+
+    Returns:
+        Dict[str, List[Dict[str, str]]]: A dictionary containing a list of email details.
+    """
     service = build("gmail", "v1", credentials=Credentials(context.authorization.token))
 
-    # Request a list of all the messages
+    try:
-    result = service.users().messages().list(userId="me").execute()
+        messages = (
-    messages = result.get("messages")
+            service.users().messages().list(userId="me").execute().get("messages", [])
+        )
 
-    # If there are no messages, return an empty string
+        if not messages:
-    if not messages:
+            return {"emails": []}
-        return ""
 
-    emails = []
+        emails = []
+        for msg in messages[:n_emails]:
+            try:
+                email_data = (
+                    service.users().messages().get(userId="me", id=msg["id"]).execute()
+                )
+                email_details = parse_email(email_data)
+                if email_details:
+                    emails.append(email_details)
+            except Exception as e:
+                print(f"Error reading email {msg['id']}: {e}")
 
-    for msg in messages[:n_emails]:
+        return json.dumps({"emails": emails})
-        txt = service.users().messages().get(userId="me", id=msg["id"]).execute()
 
-        try:
+    except Exception as e:
-            payload = txt["payload"]
+        print(f"Error reading emails: {e}")
-            headers = payload["headers"]
+        return "Error reading emails"
 
-            for d in headers:
-                if d["name"] == "From":
-                    from_ = d["value"]
-                if d["name"] == "Date":
-                    date = d["value"]
-                if d["name"] == "Subject":
-                    subject = d["value"]
-                else:
-                    subject = "No subject"
 
-            data = None
+def parse_email(email_data: Dict[str, Any]) -> Optional[Dict[str, str]]:
-            parts = payload.get("parts")
+    """
-            if parts:
+    Parse email data and extract relevant information.
-                part = parts[0]
-                body = part.get("body")
-                if body:
-                    data = body.get("data")
-                    if data:
-                        data = urlsafe_b64decode(data).decode()
 
-            email_details = {
+    Args:
-                "from": from_,
+        email_data (Dict[str, Any]): Raw email data from Gmail API.
-                "date": date,
-                "subject": subject,
-                "body": clean_email_body(data) if data else "",
-            }
-            emails.append(email_details)
 
-        except Exception as e:
+    Returns:
-            print(f"Error reading email {msg['id']}: {e}", "ERROR")
+        Optional[Dict[str, str]]: Parsed email details or None if parsing fails.
-            continue
+    """
+    try:
+        payload = email_data["payload"]
+        headers = {d["name"].lower(): d["value"] for d in payload["headers"]}
 
-    return {"emails": emails}
+        body_data = get_email_body(payload)
+
+        return {
+            "from": headers.get("from", ""),
+            "date": headers.get("date", ""),
+            "subject": headers.get("subject", "No subject"),
+            "body": clean_email_body(body_data) if body_data else "",
+        }
+    except Exception as e:
+        print(f"Error parsing email {email_data.get('id', 'unknown')}: {e}")
+        return None
+
+
+def get_email_body(payload: Dict[str, Any]) -> Optional[str]:
+    """
+    Extract email body from payload.
+
+    Args:
+        payload (Dict[str, Any]): Email payload data.
+
+    Returns:
+        Optional[str]: Decoded email body or None if not found.
+    """
+    if "body" in payload and payload["body"].get("data"):
+        return urlsafe_b64decode(payload["body"]["data"]).decode()
+
+    for part in payload.get("parts", []):
+        if part.get("mimeType") == "text/plain" and "data" in part["body"]:
+            return urlsafe_b64decode(part["body"]["data"]).decode()
+
+    return None
 
 
 def clean_email_body(body: str) -> str:
-    """Remove HTML tags and non-sentence elements from email body text."""
+    """
+    Remove HTML tags and clean up email body text while preserving most content.
 
-    # Remove HTML tags using BeautifulSoup
+    Args:
-    soup = BeautifulSoup(body, "html.parser")
+        body (str): The raw email body text.
-    text = soup.get_text(separator=" ")
 
-    # Remove any non-sentence elements (e.g., URLs, email addresses, etc.)
+    Returns:
-    text = re.sub(r"\S*@\S*\s?", "", text)  # Remove emails
+        str: Cleaned email body text.
-    text = re.sub(r"http\S+", "", text)  # Remove URLs
+    """
-    text = re.sub(r"[^.!?a-zA-Z0-9\s]", "", text)  # Remove non-sentence characters
+    try:
-    text = " ".join(text.split())  # Remove extra whitespace
+        # Remove HTML tags using BeautifulSoup
+        soup = BeautifulSoup(body, "html.parser")
+        text = soup.get_text(separator=" ")
+
+        # Clean up the text
+        text = clean_text(text)
+
+        return text.strip()
+    except Exception as e:
+        print(f"Error cleaning email body: {e}")
+        return body
+
+
+def clean_text(text: str) -> str:
+    """
+    Clean up the text while preserving most content.
+
+    Args:
+        text (str): The input text.
+
+    Returns:
+        str: Cleaned text.
+    """
+    # Replace multiple newlines with a single newline
+    text = re.sub(r"\n+", "\n", text)
+
+    # Replace multiple spaces with a single space
+    text = re.sub(r"\s+", " ", text)
+
+    # Remove leading/trailing whitespace from each line
+    text = "\n".join(line.strip() for line in text.split("\n"))
 
     return text

toolkits/slack/arcade_slack/tools/chat.py

@@ -1,16 +1,16 @@
 from typing import Annotated
-from arcade.core.errors import ToolExecutionError, RetryableToolError
+
+from slack_sdk import WebClient
+from slack_sdk.errors import SlackApiError
+
+from arcade.core.errors import RetryableToolError, ToolExecutionError
 from arcade.core.schema import ToolContext
 from arcade.sdk import tool
 from arcade.sdk.auth import SlackUser
-from slack_sdk import WebClient
-from slack_sdk.errors import SlackApiError
 
 
 @tool(
     requires_auth=SlackUser(
-        # TODO reduce this to chat:write, im:write, users.profile:read, users:read
-        # when incremental auth works
         scope=[
             "chat:write",
             "im:write",
@@ -74,8 +74,6 @@ def format_users(userListResponse: dict) -> str:
 
 @tool(
     requires_auth=SlackUser(
-        # TODO reduce this to chat:write, channels:read, groups:read
-        # when incremental auth works
         scope=[
             "chat:write",
             "im:write",