feat: per-request toolkit isolation for concurrent multi-user requests

Add clone-based isolation so each Slack/Telegram/WhatsApp request gets
fresh toolkit state (creds, service, caches) while sharing heavy
resources (DB, GoogleAuth, MCP connections) by reference.

Changes:
- Toolkit: add clone() with _clone_reset_attrs for per-user state reset
- Function: add clone_for() to rebind entrypoints to cloned toolkit
- deep_copy_field: share GoogleAuth by reference (like MCP tools),
  fall back to clone() when deepcopy fails (Google httplib2 clients)
- Interface routers: call entity.deep_copy() before arun() so each
  request runs on an isolated agent copy
- AgentOS: wire toolkit._db at init time (_initialize_agents) so the
  OAuth callback closure has DB access before any request arrives
- Cookbook: Slack Gmail agent with DB-backed OAuth flow

Depends on: feat/google-auth-db-storage

Author: Mustafa-Esoofally

cookbook/05_agent_os/interfaces/slack/gmail_agent.py

@@ -0,0 +1,101 @@
+"""
+Slack Gmail Agent (DB-backed OAuth)
+====================================
+
+Slack bot that can read/send Gmail using DB-backed OAuth tokens.
+When a user asks about email, the agent checks the DB for stored tokens.
+If none exist, it returns an OAuth URL the user clicks to authenticate.
+
+Setup:
+  1. Google Cloud: Create OAuth 2.0 Web Application credentials
+     - Authorized redirect URI: https://<your-ngrok>/google/oauth/callback
+  2. Set env vars:
+     - GOOGLE_CLIENT_ID, GOOGLE_CLIENT_SECRET
+     - GOOGLE_REDIRECT_URI=https://<your-ngrok>/google/oauth/callback
+     - SLACK_BOT_TOKEN, SLACK_SIGNING_SECRET
+     - OPENAI_API_KEY
+  3. Start pgvector: ./cookbook/scripts/run_pgvector.sh
+  4. Start ngrok:    ngrok http 7777
+  5. Run:            .venvs/demo/bin/python cookbook/05_agent_os/interfaces/slack/gmail_agent.py
+
+Flow:
+  User: "What are my latest emails?"
+  Bot:  Returns OAuth URL (first time) -> user clicks -> Google consent -> callback saves token
+  User: Retries -> Bot reads Gmail via stored token
+"""
+
+from agno.agent import Agent
+from agno.db.postgres.postgres import PostgresDb
+from agno.models.openai import OpenAIChat
+from agno.os.app import AgentOS
+from agno.os.interfaces.slack import Slack
+from agno.tools.google.auth import GoogleAuth
+from agno.tools.google.gmail import GmailTools
+
+# ---------------------------------------------------------------------------
+# Database (shared between agent session storage and OAuth token storage)
+# ---------------------------------------------------------------------------
+
+db = PostgresDb(
+    db_url="postgresql+psycopg://ai:ai@localhost:5532/ai",
+)
+
+# ---------------------------------------------------------------------------
+# Google Auth + Gmail
+# ---------------------------------------------------------------------------
+
+google_auth = GoogleAuth()
+
+gmail = GmailTools(
+    google_auth=google_auth,
+    get_latest_emails=True,
+    get_emails_from_user=True,
+    get_unread_emails=True,
+    search_emails=True,
+    send_email=True,
+    create_draft_email=True,
+)
+
+# ---------------------------------------------------------------------------
+# Agent
+# ---------------------------------------------------------------------------
+
+gmail_agent = Agent(
+    name="Gmail Agent",
+    model=OpenAIChat(id="gpt-4o"),
+    tools=[google_auth, gmail],
+    db=db,
+    add_history_to_context=True,
+    num_history_runs=3,
+    add_datetime_to_context=True,
+    instructions=[
+        "You are a helpful email assistant connected to Gmail.",
+        "When authentication is needed, share the OAuth URL and ask the user to click it.",
+        "After they authenticate, retry the original request.",
+    ],
+)
+
+# ---------------------------------------------------------------------------
+# AgentOS + Slack + OAuth callback
+# ---------------------------------------------------------------------------
+
+agent_os = AgentOS(
+    agents=[gmail_agent],
+    interfaces=[
+        Slack(
+            agent=gmail_agent,
+            reply_to_mentions_only=True,
+            resolve_user_identity=True,
+        )
+    ],
+)
+
+app = agent_os.get_app()
+
+# Attach the OAuth callback endpoint to the same FastAPI app
+# Google redirects here after user consents: /google/oauth/callback?code=...&state=...
+app.include_router(google_auth.get_oauth_router())
+
+
+if __name__ == "__main__":
+    agent_os.serve(app="gmail_agent:app", reload=True)

libs/agno/agno/agent/_utils.py

@@ -171,18 +171,20 @@ def deep_copy_field(agent: Agent, field_name: str, field_value: Any) -> Any:
             copied_tools = []
             for tool in field_value:  # type: ignore
                 try:
-                    # Share MCP tools (they maintain server connections)
-                    is_mcp_tool = hasattr(type(tool), "__mro__") and any(
-                        c.__name__ in ["MCPTools", "MultiMCPTools"] for c in type(tool).__mro__
-                    )
-                    if is_mcp_tool:
+                    # Share tools that maintain connections or act as coordinators
+                    mro_names = {c.__name__ for c in type(tool).__mro__} if hasattr(type(tool), "__mro__") else set()
+                    is_shared_tool = bool(mro_names & {"MCPTools", "MultiMCPTools", "GoogleAuth"})
+                    if is_shared_tool:
                         copied_tools.append(tool)
                     else:
                         try:
                             copied_tools.append(deepcopy(tool))
                         except Exception:
-                            # Tool can't be deep copied, share by reference
-                            copied_tools.append(tool)
+                            # deepcopy failed (e.g. Google httplib2 clients) — try clone()
+                            if hasattr(tool, "clone") and callable(tool.clone):
+                                copied_tools.append(tool.clone())
+                            else:
+                                copied_tools.append(tool)
                 except Exception:
                     # MCP detection failed, share tool by reference to be safe
                     copied_tools.append(tool)

libs/agno/agno/os/app.py

@@ -67,6 +67,7 @@
 from agno.registry import Registry
 from agno.remote.base import RemoteDb, RemoteKnowledge
 from agno.team import RemoteTeam, Team
+from agno.tools.toolkit import Toolkit
 from agno.utils.log import log_debug, log_error, log_info, log_warning
 from agno.utils.string import generate_id, generate_id_from_name
 from agno.workflow import RemoteWorkflow, Workflow
@@ -527,6 +528,17 @@ def _initialize_agents(self) -> None:
 
             agent.initialize_agent()
 
+            # Wire agent.db to toolkits that declare _db (e.g. GoogleAuth for OAuth token storage).
+            # Must happen here (init time) so the OAuth callback closure captures a wired instance.
+            # Only sync DBs that override get_auth_token — skips async and unsupported backends.
+            if agent.db is not None and isinstance(agent.tools, list):
+                from agno.db.base import BaseDb
+
+                if isinstance(agent.db, BaseDb) and type(agent.db).get_auth_token is not BaseDb.get_auth_token:
+                    for tool in agent.tools:
+                        if isinstance(tool, Toolkit) and hasattr(tool, "_db") and tool._db is None:
+                            tool._db = agent.db
+
             # Required for the built-in routes to work
             agent.store_events = True
 
@@ -556,6 +568,18 @@ def _initialize_teams(self) -> None:
                     if isinstance(member, Agent):
                         member.team_id = None
                         member.initialize_agent()
+                        # Wire DB to member agent toolkits — same guard as _initialize_agents
+                        member_db = member.db or team.db
+                        if member_db is not None and isinstance(member.tools, list):
+                            from agno.db.base import BaseDb
+
+                            if (
+                                isinstance(member_db, BaseDb)
+                                and type(member_db).get_auth_token is not BaseDb.get_auth_token
+                            ):
+                                for tool in member.tools:
+                                    if isinstance(tool, Toolkit) and hasattr(tool, "_db") and tool._db is None:
+                                        tool._db = member_db
                     elif isinstance(member, Team):
                         member.initialize_team()
 
@@ -823,7 +847,6 @@ async def general_exception_handler(_: Request, exc: Exception) -> JSONResponse:
         # This allows middleware (like JWT) to access these values
         fastapi_app.state.agent_os_id = self.id
         fastapi_app.state.cors_allowed_origins = self.cors_allowed_origins
-
         # Store internal service token for scheduler auth bypass
         if self._internal_service_token:
             fastapi_app.state.internal_service_token = self._internal_service_token

libs/agno/agno/os/interfaces/slack/router.py

@@ -191,6 +191,9 @@ async def _process_slack_event(data: dict):
             if skipped:
                 notice = "[Skipped files: " + ", ".join(skipped) + "]"
                 message_text = f"{notice}\n{message_text}"
+            # Per-request clone isolates mutable toolkit state (creds, service) between users
+            _entity = entity.deep_copy() if hasattr(entity, "deep_copy") else entity
+
             run_kwargs: Dict[str, Any] = {
                 "user_id": resolved_user_id,
                 "session_id": session_id,
@@ -207,7 +210,7 @@ async def _process_slack_event(data: dict):
                 "audio": audio or None,
             }
 
-            response = await entity.arun(message_text, **run_kwargs)  # type: ignore[union-attr]
+            response = await _entity.arun(message_text, **run_kwargs)  # type: ignore[union-attr]
 
             if response:
                 if response.status == "ERROR":
@@ -312,6 +315,8 @@ async def _stream_slack_response(data: dict):
             if skipped:
                 notice = "[Skipped files: " + ", ".join(skipped) + "]"
                 message_text = f"{notice}\n{message_text}"
+            _entity = entity.deep_copy() if hasattr(entity, "deep_copy") else entity
+
             run_kwargs: Dict[str, Any] = {
                 "stream": True,
                 # Enables event-level chunks for task card and tool lifecycle rendering
@@ -331,7 +336,7 @@ async def _stream_slack_response(data: dict):
                 "audio": audio or None,
             }
 
-            response_stream = entity.arun(message_text, **run_kwargs)  # type: ignore[union-attr]
+            response_stream = _entity.arun(message_text, **run_kwargs)  # type: ignore[union-attr]
 
             if response_stream is None:
                 try:

libs/agno/agno/os/interfaces/telegram/router.py

@@ -204,6 +204,8 @@ async def _stream_response(
         message_thread_id: Optional[int],
         is_private: bool = False,
     ) -> None:
+        _entity = entity.deep_copy() if hasattr(entity, "deep_copy") else entity
+
         is_workflow = entity_type == "workflow"
         stream_kwargs: dict = dict(stream=True, stream_events=True, **run_kwargs)
         if not is_workflow:
@@ -219,7 +221,7 @@ async def _stream_response(
         )
 
         try:
-            async for event in entity.arun(message_text, **stream_kwargs):  # type: ignore[union-attr]
+            async for event in _entity.arun(message_text, **stream_kwargs):  # type: ignore[union-attr]
                 if isinstance(event, (RunOutput, TeamRunOutput)):
                     state.final_run_output = event
                     continue
@@ -261,7 +263,8 @@ async def _sync_response(
         reply_to: Optional[int],
         message_thread_id: Optional[int],
     ) -> None:
-        response = await entity.arun(message_text, **run_kwargs)  # type: ignore[union-attr]
+        _entity = entity.deep_copy() if hasattr(entity, "deep_copy") else entity
+        response = await _entity.arun(message_text, **run_kwargs)  # type: ignore[union-attr]
         if not response or response.status == "ERROR":
             if response:
                 log_error(response.content)

libs/agno/agno/os/interfaces/whatsapp/router.py

@@ -313,9 +313,10 @@ async def _keep_typing():
                 except asyncio.CancelledError:
                     pass
 
+            _entity = entity.deep_copy() if hasattr(entity, "deep_copy") else entity
             typing_task = asyncio.create_task(_keep_typing())
             try:
-                response = await entity.arun(parsed.text, **run_kwargs)  # type: ignore[union-attr]
+                response = await _entity.arun(parsed.text, **run_kwargs)  # type: ignore[union-attr]
             finally:
                 typing_task.cancel()
 

libs/agno/agno/team/_utils.py

@@ -183,18 +183,20 @@ def _deep_copy_field(team: Team, field_name: str, field_value: Any) -> Any:
             copied_tools = []
             for tool in field_value:
                 try:
-                    # Share MCP tools (they maintain server connections)
-                    is_mcp_tool = hasattr(type(tool), "__mro__") and any(
-                        c.__name__ in ["MCPTools", "MultiMCPTools"] for c in type(tool).__mro__
-                    )
-                    if is_mcp_tool:
+                    # Share tools that maintain connections or act as coordinators
+                    mro_names = {c.__name__ for c in type(tool).__mro__} if hasattr(type(tool), "__mro__") else set()
+                    is_shared_tool = bool(mro_names & {"MCPTools", "MultiMCPTools", "GoogleAuth"})
+                    if is_shared_tool:
                         copied_tools.append(tool)
                     else:
                         try:
                             copied_tools.append(deepcopy(tool))
                         except Exception:
-                            # Tool can't be deep copied, share by reference
-                            copied_tools.append(tool)
+                            # deepcopy failed (e.g. Google httplib2 clients) — try clone()
+                            if hasattr(tool, "clone") and callable(tool.clone):
+                                copied_tools.append(tool.clone())
+                            else:
+                                copied_tools.append(tool)
                 except Exception:
                     # MCP detection failed, share tool by reference to be safe
                     copied_tools.append(tool)
@@ -216,7 +218,6 @@ def _deep_copy_field(team: Team, field_name: str, field_value: Any) -> Any:
         "session_summary_manager",
         "compression_manager",
         "learning",
-        "skills",
     ):
         return field_value
 

libs/agno/agno/tools/function.py

@@ -234,6 +234,21 @@ def from_dict(cls, data: Dict[str, Any]) -> "Function":
             approval_type=data.get("approval_type"),
         )
 
+    def clone_for(self, new_owner: Any) -> "Function":
+        """Shallow-copy this Function with its entrypoint rebound to new_owner.
+
+        Used by Toolkit.clone() so tool methods operate on the cloned toolkit's
+        state instead of the original's. Only rebinds bound methods (__self__);
+        closures and standalone functions are left as-is.
+        """
+        import types
+
+        clone = self.model_copy()
+        ep = self.entrypoint
+        if ep is not None and hasattr(ep, "__self__"):
+            clone.entrypoint = types.MethodType(ep.__func__, new_owner)  # type: ignore[attr-defined]
+        return clone
+
     def model_copy(self, *, deep: bool = False) -> "Function":
         """
         Override model_copy to handle callable fields that can't be deep copied (pickled).

libs/agno/agno/tools/toolkit.py

@@ -12,6 +12,10 @@ class Toolkit:
     # When True, the Agent will automatically call connect() before using tools and close() after
     _requires_connect: bool = False
 
+    # Subclasses override with attr names that hold per-user mutable state (creds, service, etc.).
+    # clone() resets these to None so each request rebuilds them for the current user.
+    _clone_reset_attrs: tuple = ()
+
     def __init__(
         self,
         name: str = "toolkit",
@@ -345,6 +349,30 @@ def close(self) -> None:
         """
         pass
 
+    def clone(self) -> "Toolkit":
+        """Shallow-copy this toolkit and reset per-user mutable state.
+
+        Used by deep_copy_field() as a fallback when deepcopy() fails (e.g. Google
+        API clients hold unpicklable httplib2 transports). Config attrs (scopes,
+        credentials_path, etc.) are shared; mutable attrs listed in _clone_reset_attrs
+        are set to None so the auth decorator rebuilds them for the current user.
+        """
+        import copy
+
+        clone = copy.copy(self)
+        for attr in self._clone_reset_attrs:
+            setattr(clone, attr, None)
+
+        # Rebuild function dicts so entrypoints are bound to the clone, not the original
+        clone.functions = OrderedDict()
+        clone.async_functions = OrderedDict()
+        for name, fn in self.functions.items():
+            clone.functions[name] = fn.clone_for(clone)
+        for name, fn in self.async_functions.items():
+            clone.async_functions[name] = fn.clone_for(clone)
+
+        return clone
+
     def _check_path(self, file_name: str, base_dir: Path, restrict_to_base_dir: bool = True) -> Tuple[bool, Path]:
         """Check if the file path is within the base directory.