Remove ClickUpAgent implementation and instructions; add NotionProjectAgent with instructions and tools

agency_ai_demo/agents/ClickUpAgent/__init__.py

@@ -1 +0,0 @@
-from .ClickUpAgent import ClickUpAgent

agency_ai_demo/agents/ClickUpAgent/instructions.md

@@ -1,2 +0,0 @@
-# ClickUpAgent Instructions
-

agency_ai_demo/agents/ClickUpAgent/tools/ClickUpTools.py

@@ -1,1191 +0,0 @@
-import os
-import sys
-import requests
-from dotenv import load_dotenv
-from langchain.tools import StructuredTool, BaseTool, tool
-from langchain_core.pydantic_v1 import BaseModel, Field
-from langchain_core.tools import ToolException
-from pydantic import ValidationError
-from typing import Any, Type, List, Dict
-import datetime
-import json
-import re
-from composio.tools.local.clickup.actions.base import OpenAPIAction
-from composio.tools.local.clickup.actions.create_task import (
-    CreateTask,
-    CreateTaskResponse,
-)
-from composio.tools.local.clickup.actions.delete_task import (
-    DeleteTask,
-    DeleteTaskRequest,
-    DeleteTaskResponse,
-)
-from composio.tools.local.clickup.actions.update_task import (
-    UpdateTask,
-    UpdateTaskRequest,
-    UpdateTaskResponse,
-)
-from composio.tools.local.clickup.actions.add_dependency import (
-    AddDependency,
-    AddDependencyRequest,
-    AddDependencyResponse,
-)
-from composio.tools.local.clickup.actions.get_list import (
-    GetList,
-    GetListRequest,
-    GetListResponse,
-)
-from composio.tools.local.clickup.actions.get_tasks import (
-    GetTasks,
-    GetTasksRequest,
-    GetTasksResponse,
-)
-from composio.tools.local.clickup.actions.get_task import (
-    GetTask,
-    GetTaskRequest,
-    GetTaskResponse,
-)
-
-# Add the parent directory to sys.path to enable imports
-sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
-from utils.tool_wrapper import convert_langchain_tools
-from utils.ensure_serializable import ensure_serializable
-
-load_dotenv()
-CLICKUP_TOKEN = os.getenv("CLICKUP_TOKEN")
-
-
-class CreateTaskSchema(BaseModel):
-    """Request schema for `CreateTask`"""
-
-    list_id: int = Field(
-        ...,
-        alias="list_id",
-        description="",
-    )
-    custom_task_ids: bool = Field(
-        default=None,
-        alias="custom_task_ids",
-        description=(
-            'If you want to reference a task by it"s custom task id, this value must '
-            "be `true`. "
-        ),
-    )
-    team_id: int = Field(
-        default=None,
-        alias="team_id",
-        description=(
-            "Only used when the `custom_task_ids` parameter is set to `true`.   For example: "
-            "`custom_task_ids=true&team_id=123`. "
-        ),
-    )
-    tags: List[str] = Field(
-        default=None,
-        alias="tags",
-        description="",
-    )
-    description: str = Field(
-        default=None,
-        alias="description",
-        description="Description",
-    )
-    name: str = Field(
-        default=...,
-        alias="name",
-        description="Name",
-    )
-    assignees: List[int] = Field(
-        ...,
-        alias="assignees",
-        description="To create tasks in ClickUp, the list of assignees is mandatory; NEVER assign the responsible parties on your own; the user must explicitly inform who the responsible parties are.",
-        examples=[18951490, 48772077],
-    )
-    status: str = Field(
-        default=None,
-        alias="status",
-        description="Status",
-    )
-    priority: int = Field(
-        default=None,
-        alias="priority",
-        description="Priority",
-    )
-    due_date: int = Field(
-        default=None,
-        alias="due_date",
-        description="Due Date",
-    )
-    due_date_time: bool = Field(
-        default=None,
-        alias="due_date_time",
-        description="Due Date Time",
-    )
-    time_estimate: int = Field(
-        default=None,
-        alias="time_estimate",
-        description="Time Estimate",
-    )
-    start_date: int = Field(
-        default=None,
-        alias="start_date",
-        description="Start Date",
-    )
-    start_date_time: bool = Field(
-        default=None,
-        alias="start_date_time",
-        description="Start Date Time",
-    )
-    notify_all: bool = Field(
-        default=None,
-        alias="notify_all",
-        description=(
-            "If `notify_all` is true, notifications will be sent to everyone including "
-            "the creator of the comment. "
-        ),
-    )
-    parent: str = Field(
-        default=None,
-        alias="parent",
-        description=(
-            "You can create a subtask by including an existing task ID.   The `parent` "
-            "task ID you include cannot be a subtask, and must be in the same List specified "
-            "in the path parameter. "
-        ),
-    )
-    links_to: str = Field(
-        default=None,
-        alias="links_to",
-        description="Include a task ID to create a linked dependency with your new task.",
-    )
-    check_required_custom_fields: bool = Field(
-        default=None,
-        alias="check_required_custom_fields",
-        description=(
-            "When creating a task via API any required Custom Fields are ignored by default "
-            "(`false`).   You can enforce required Custom Fields by including `check_required_custom_fields: "
-            "true`. "
-        ),
-    )
-    custom_fields: List[dict] = Field(
-        default=None,
-        alias="custom_fields",
-        description="[Filter by Custom Fields.](https://clickup.com/api)",
-    )
-    custom_item_id: int = Field(
-        default=None,
-        alias="custom_item_id",
-        description=(
-            'To create a task that doesn"t use a custom task type, either don"t include '
-            'this field in the request body, or send `"null"`.    To create this task '
-            "as a Milestone, send a value of `1`.   To use a custom task type, send the "
-            "custom task type ID as defined in your Workspace, such as `2`. "
-        ),
-    )
-
-
-class CreateTaskTool(BaseTool):
-    name: str = "create_task_tool"
-    description: str = """
-    Tool to create a new task in ClickUp based on the provided parameters.
-    - Create Task:
-        Invoke: "CreateTaskTool" with the appropriate parameters.
-
-    Parameters:
-    - list_id (required): The ID of the list to create the task in. Example: 901307715461
-    - name (required): The name of the task
-    - assignees (required): List of user IDs to assign to the task
-    
-    IMPORTANT
-    - Always use 'date_to_timestamp' tool to convert dates from 'YYYY-MM-DD' to Unix millisecond timestamps before setting dates on ClickUp
-    """
-    args_schema: Type[BaseModel] = CreateTaskSchema
-    headers: dict = {"Authorization": f"{CLICKUP_TOKEN}"}
-
-    def __init__(self, **data):
-        super().__init__(**data)
-
-    def _run(self, **kwargs) -> Any:
-        """Executes task creation in ClickUp"""
-
-        # Log the received parameters to help debug
-        print("\n==== CreateTaskTool._run received parameters: ====")
-        print(f"kwargs: {kwargs}")
-
-        # Extract list_id from different possible locations
-        list_id = None
-
-        # 1. Direct list_id parameter
-        if "list_id" in kwargs:
-            list_id = kwargs.get("list_id")
-            print(f"Found list_id in direct parameter: {list_id}")
-
-        # 2. Check if list_id is inside nested kwargs
-        elif "kwargs" in kwargs and isinstance(kwargs["kwargs"], dict):
-            list_id = kwargs["kwargs"].get("list_id")
-            print(f"Found list_id in nested kwargs: {list_id}")
-
-        # 3. Check if list_id is in a string format in any parameter
-        for k, v in kwargs.items():
-            if isinstance(v, str) and v.isdigit():
-                try:
-                    list_id = int(v)
-                    print(f"Found list_id in parameter {k}: {list_id}")
-                    break
-                except ValueError:
-                    pass
-            elif isinstance(v, str) and "901307715461" in v:
-                list_id = 901307715461
-                print(f"Found list_id in parameter {k}: {list_id}")
-                break
-
-        # 4. Hardcoded fallback for this specific test case
-        if not list_id:
-            print("No list_id found in parameters, using hardcoded value 901307715461")
-            list_id = 901307715461
-
-        print(f"list_id being used: {list_id}")
-        print("==== End parameters ====\n")
-
-        action = CreateTask()
-
-        url = f"{action.url}{action.path}".format(list_id=list_id)
-        print(f"URL being used: {url}")
-
-        # Make sure all parameters are JSON serializable and extract from kwargs if needed
-        params = {}
-
-        # If name is not directly in kwargs, try to find it
-        if (
-            "name" not in kwargs
-            and "kwargs" in kwargs
-            and isinstance(kwargs["kwargs"], dict)
-        ):
-            for k, v in kwargs["kwargs"].items():
-                if k == "name" or (isinstance(v, str) and "API TEST TASK" in v):
-                    params["name"] = "API TEST TASK"
-                    break
-
-        # If assignees is not directly in kwargs, try to find it
-        if (
-            "assignees" not in kwargs
-            and "kwargs" in kwargs
-            and isinstance(kwargs["kwargs"], dict)
-        ):
-            for k, v in kwargs["kwargs"].items():
-                if k == "assignees" or (isinstance(v, str) and "81918955" in v):
-                    params["assignees"] = [81918955]
-                    break
-
-        # Add any other parameters from kwargs
-        for key, value in kwargs.items():
-            if value is not None and key != "kwargs" and key != "list_id":
-                params[key] = ensure_serializable(value)
-
-        # For testing, ensure we have the minimum required parameters
-        if "name" not in params:
-            params["name"] = "API TEST TASK"
-
-        if "assignees" not in params:
-            params["assignees"] = [81918955]
-
-        print(f"Request parameters: {params}")
-
-        response = requests.post(url, headers=self.headers, json=params)
-        print(f"Response status code: {response.status_code}")
-
-        if response.status_code == 201:
-            response_json = response.json()
-        else:
-            try:
-                response_json = response.json()
-                print(f"Error response: {response_json}")
-            except requests.JSONDecodeError:
-                response_json = {"error": "Invalid JSON response"}
-                print("Could not decode JSON response")
-
-        response = CreateTaskResponse(data=response_json)
-        filtered_response = {
-            "id": response.data.get("id"),
-            "name": response.data.get("name"),
-            "status": response.data.get("status", {}).get("status"),
-            "assignees": response.data.get("assignees"),
-            "due_date": response.data.get("due_date"),
-            "error": response.data.get("err"),
-        }
-
-        print(f"Returning filtered response: {json.dumps(filtered_response, indent=2)}")
-
-        return filtered_response
-
-
-class DeleteTaskTool(BaseTool):
-    name: str = "delete_task_tool"
-    description: str = """
-    Tool to delete a task in ClickUp based on its ID.
-    - Delete Task:
-        Invoke: "DeleteTaskTool" with the appropriate parameters.
-        
-    Parameters:
-    - task_id (required): The ID of the task to delete
-    """
-
-    args_schema: Type[BaseModel] = DeleteTaskRequest
-    headers: dict = {"Authorization": f"{CLICKUP_TOKEN}"}
-
-    def __init__(self, **data):
-        super().__init__(**data)
-
-    def _run(self, **kwargs) -> Any:
-        """Executes a task deletion in ClickUp"""
-
-        # Log the received parameters to help debug
-        print("\n==== DeleteTaskTool._run received parameters: ====")
-        print(f"kwargs: {kwargs}")
-
-        # Extract task_id from different possible locations
-        task_id = None
-        task_name = None
-
-        # 1. Direct task_id parameter
-        if "task_id" in kwargs:
-            task_id = kwargs.get("task_id")
-            print(f"Found task_id in direct parameter: {task_id}")
-
-        # 2. Check if task_id is inside nested kwargs
-        if "kwargs" in kwargs and isinstance(kwargs["kwargs"], dict):
-            task_id = task_id or kwargs["kwargs"].get("task_id")
-            print(f"Found task_id in nested kwargs: {task_id}")
-
-        # 3. Check if task_id is in FieldInfo format
-        if "kwargs" in kwargs and hasattr(kwargs["kwargs"], "task_id"):
-            if hasattr(kwargs["kwargs"].task_id, "default"):
-                task_id = kwargs["kwargs"].task_id.default
-                print(f"Found task_id in FieldInfo default: {task_id}")
-
-        # 4. Check for task_id in description or raw query
-        if "kwargs" in kwargs and hasattr(kwargs["kwargs"], "description"):
-            desc = kwargs["kwargs"].description
-            # Look for task ID pattern in the description
-            task_id_match = re.search(r'task_id[=:]\s*["\']?([0-9a-z]{8,})["\']?', desc)
-            if task_id_match:
-                task_id = task_id_match.group(1)
-                print(f"Found task_id in description: {task_id}")
-
-            # Look for task name in the description
-            task_name_match = re.search(r'task\s+["\']([^"\']+)["\']', desc)
-            if task_name_match:
-                task_name = task_name_match.group(1).strip()
-                print(f"Found task_name in description: {task_name}")
-
-        # 5. Check any string parameters for task_id
-        for k, v in kwargs.items():
-            if isinstance(v, str):
-                # Check if the parameter contains a task ID pattern
-                task_id_match = re.search(
-                    r'task_id[=:]\s*["\']?([0-9a-z]{8,})["\']?', v
-                )
-                if task_id_match:
-                    task_id = task_id_match.group(1)
-                    print(f"Found task_id in string parameter: {task_id}")
-                    break
-
-                # Check for task name pattern in the string
-                task_name_match = re.search(r'task\s+["\']([^"\']+)["\']', v)
-                if task_name_match:
-                    task_name = task_name_match.group(1).strip()
-                    print(f"Found task_name in string parameter: {task_name}")
-                    break
-
-        # 6. If task name found but no ID, try to lookup ID by name
-        if not task_id and task_name:
-            try:
-                # Get all tasks in the list to find the task ID by name
-                get_tasks_tool = GetTasksTool()
-                tasks = get_tasks_tool._run(list_id=901307715461)
-
-                # Find the task by name
-                for task in tasks:
-                    if task.get("name") == task_name:
-                        task_id = task.get("id")
-                        print(f"Found task_id {task_id} for task name '{task_name}'")
-                        break
-            except Exception as e:
-                print(f"Error getting task ID from name: {e}")
-
-        # 7. Hardcoded fallback for testing
-        if not task_id and task_name:
-            if task_name == "TEST TASK 2":
-                task_id = "86a702gha"  # Known ID of TEST TASK 2
-                print(f"Using hardcoded task_id for 'TEST TASK 2': {task_id}")
-            elif task_name == "TEST TASK":
-                task_id = "86a700c6e"  # Known ID of TEST TASK
-                print(f"Using hardcoded task_id for 'TEST TASK': {task_id}")
-
-        if not task_id:
-            raise ToolException("task_id is required for deleting a task")
-
-        print(f"task_id being used: {task_id}")
-        print("==== End parameters ====\n")
-
-        action = DeleteTask()
-
-        url = f"{action.url}{action.path}".format(task_id=task_id)
-        print(f"URL being used: {url}")
-
-        # Make sure all parameters are JSON serializable
-        params = {
-            key: ensure_serializable(value)
-            for key, value in kwargs.items()
-            if value is not None and key != "kwargs" and key != "task_id"
-        }
-
-        response = requests.delete(url, headers=self.headers, params=params)
-        print(f"Response status code: {response.status_code}")
-
-        if response.status_code == 200:
-            response_json = response.json()
-        else:
-            try:
-                response_json = response.json()
-                print(f"Error response: {response_json}")
-            except requests.JSONDecodeError:
-                response_json = {"error": "Invalid JSON response"}
-                print("Could not decode JSON response")
-
-        response = DeleteTaskResponse(data=response_json)
-
-        result_message = f"Task '{task_name or task_id}' successfully deleted"
-
-        if "err" in response.data:
-            result_message = f"Error: {response.data['err']}"
-
-        print(f"Result: {result_message}")
-
-        return result_message
-
-
-class CustomUpdateTaskRequest(BaseModel):
-    """Request schema for `UpdateTask`"""
-
-    task_id: str = Field(
-        ...,
-        alias="task_id",
-        description="",
-    )
-    custom_task_ids: bool = Field(
-        default=None,
-        alias="custom_task_ids",
-        description=(
-            'If you want to reference a task by it"s custom task id, this value must '
-            "be `true`. "
-        ),
-    )
-    team_id: int = Field(
-        default=None,
-        alias="team_id",
-        description=(
-            "Only used when the `custom_task_ids` parameter is set to `true`.   For example: "
-            "`custom_task_ids=true&team_id=123`. "
-        ),
-    )
-    description: str = Field(
-        default=None,
-        alias="description",
-        description='To clear the task description, include `Description` with `" "`.',
-    )
-    custom_item_id: int = Field(
-        default=None,
-        alias="custom_item_id",
-        description=(
-            'To convert an item using a custom task type into a task, send `"null"`. '
-            "   To update this task to be a Milestone, send a value of `1`.    To use "
-            "a custom task type, send the custom task type ID as defined in your Workspace, "
-            "such as `2`. "
-        ),
-    )
-    name: str = Field(
-        default=None,
-        alias="name",
-        description="Name",
-    )
-    status: str = Field(
-        default=None,
-        alias="status",
-        description="Status",
-    )
-    priority: int = Field(
-        default=None,
-        alias="priority",
-        description="Priority",
-    )
-    due_date: int = Field(
-        default=None,
-        alias="due_date",
-        description="Due Date in Unix millisecond timestamps",
-    )
-    due_date_time: bool = Field(
-        default=None,
-        alias="due_date_time",
-        description="Due Date Time",
-    )
-    parent: str = Field(
-        default=None,
-        alias="parent",
-        description=(
-            'You can move a subtask to another parent task by including `"parent"` with '
-            'a valid `task id`.   You cannot convert a subtask to a task by setting `"parent"` '
-            "to `null`. "
-        ),
-    )
-    time_estimate: int = Field(
-        default=None,
-        alias="time_estimate",
-        description="Time Estimate",
-    )
-    start_date: int = Field(
-        default=None,
-        alias="start_date",
-        description="Start Date in Unix millisecond timestamps",
-    )
-    start_date_time: bool = Field(
-        default=None,
-        alias="start_date_time",
-        description="Start Date Time",
-    )
-    assignees: List[Dict[str, List[int]]] = Field(
-        default=[{"add": [], "rem": []}],
-        description="List of user IDs to add or remove as assignees",
-        examples=[{"add": [81918955, 82061927], "rem": [18951490, 48772077]}],
-    )
-    archived: bool = Field(
-        default=None,
-        alias="archived",
-        description="Archived",
-    )
-
-
-class CustomUpdateTask(OpenAPIAction):
-    """Update a task by including one or more fields in the request body."""
-
-    _tags = ["Tasks"]
-    _display_name = "update_task"
-    _request_schema = CustomUpdateTaskRequest
-    _response_schema = CustomUpdateTaskRequest
-
-    url = "https://api.clickup.com/api/v2"
-    path = "/task/{task_id}"
-    method = "put"
-    operation_id = "Tasks_updateTaskFields"
-    action_identifier = "/task/{task_id}_put"
-
-    path_params = {"task_id": "task_id"}
-    query_params = {"custom_task_ids": "custom_task_ids", "team_id": "team_id"}
-    header_params = {}
-    request_params = {
-        "description": {"__alias": "description"},
-        "custom_item_id": {"__alias": "custom_item_id"},
-        "name": {"__alias": "name"},
-        "status": {"__alias": "status"},
-        "priority": {"__alias": "priority"},
-        "due_date": {"__alias": "due_date"},
-        "due_date_time": {"__alias": "due_date_time"},
-        "parent": {"__alias": "parent"},
-        "time_estimate": {"__alias": "time_estimate"},
-        "start_date": {"__alias": "start_date"},
-        "start_date_time": {"__alias": "start_date_time"},
-        "assignees": {"__alias": "assignees"},
-        "archived": {"__alias": "archived"},
-    }
-
-
-class UpdateTaskTool(BaseTool):
-    name: str = "update_task_tool"
-    description: str = """
-    Tool to update a task in ClickUp based on the provided parameters.
-    - Update Task:
-        Invoke: "UpdateTaskTool" with the appropriate parameters.
-        
-    Parameters:
-    - task_id (required): The ID of the task to update
-    - name (optional): New name for the task
-    - status (optional): New status for the task
-    
-    IMPORTANT
-    - Always use 'date_to_timestamp' tool to convert dates from 'YYYY-MM-DD' to Unix millisecond timestamps when setting dates on ClickUp
-    """
-    args_schema: Type[BaseModel] = CustomUpdateTaskRequest
-    headers: dict = {"Authorization": f"{CLICKUP_TOKEN}"}
-
-    def __init__(self, **data):
-        super().__init__(**data)
-
-    def _run(self, **kwargs) -> Any:
-        """Executes task update in ClickUp"""
-
-        # Log the received parameters to help debug
-        print("\n==== UpdateTaskTool._run received parameters: ====")
-        print(f"kwargs: {kwargs}")
-
-        # Extract task_id from different possible locations
-        task_id = None
-        update_params = {}
-        task_name_to_update = None
-
-        # 1. Direct task_id parameter
-        if "task_id" in kwargs:
-            task_id = kwargs.get("task_id")
-            print(f"Found task_id in direct parameter: {task_id}")
-
-        # 2. Check if task_id is inside nested kwargs
-        elif "kwargs" in kwargs and isinstance(kwargs["kwargs"], dict):
-            task_id = kwargs["kwargs"].get("task_id")
-            print(f"Found task_id in nested kwargs: {task_id}")
-
-        # 3. Check if there's a task_id in the kwargs object of FieldInfo type
-        elif "kwargs" in kwargs and hasattr(kwargs["kwargs"], "default"):
-            # Try to parse it from the description if it contains the task ID
-            if (
-                hasattr(kwargs["kwargs"], "description")
-                and kwargs["kwargs"].description
-            ):
-                desc = kwargs["kwargs"].description
-                # Look for common task ID patterns (alphanumeric with at least 8 chars)
-                task_id_match = re.search(r"(86a[0-9a-z]{5,})", desc)
-                if task_id_match:
-                    task_id = task_id_match.group(1)
-                    print(f"Found task_id in FieldInfo description: {task_id}")
-
-        # 4. Look for task name to update in parameters
-        for k, v in kwargs.items():
-            if isinstance(v, str):
-                # Check if it looks like a task ID (alphanumeric pattern)
-                if re.match(r"^[0-9a-z]{8,}$", v):
-                    task_id = v
-                    print(f"Found task_id in parameter {k}: {task_id}")
-                    break
-
-                # Look for patterns like "Change 'TEST TASK 2' to 'TEST TASK 1000'"
-                change_pattern = re.search(
-                    r"Change\s+['\"]?(.*?)['\"]?\s+to\s+['\"]?(.*?)['\"]?", v
-                )
-                if change_pattern:
-                    task_name_to_update = change_pattern.group(1).strip()
-                    new_name = change_pattern.group(2).strip()
-                    update_params["name"] = new_name
-                    print(
-                        f"Found task to update: '{task_name_to_update}' to '{new_name}'"
-                    )
-                    break
-
-                # If string contains task names, extract them
-                elif "TEST TASK" in v:
-                    if "TEST TASK 2" in v:
-                        task_name_to_update = "TEST TASK 2"
-                    else:
-                        task_name_to_update = "TEST TASK"
-
-                    # Look for new name in the string
-                    name_pattern = re.search(r"to\s+['\"]?(.*?)['\"]?(?:\s|$)", v)
-                    if name_pattern:
-                        new_name = name_pattern.group(1).strip()
-                        update_params["name"] = new_name
-                        print(
-                            f"Found task to update: '{task_name_to_update}' to '{new_name}'"
-                        )
-
-        # 5. If we have a task name but no ID, look up the ID
-        if not task_id and task_name_to_update:
-            try:
-                # Get all tasks in the list to find the task ID by name
-                get_tasks_tool = GetTasksTool()
-                tasks = get_tasks_tool._run(list_id=901307715461)
-                # Find the task by name
-                for task in tasks:
-                    if task.get("name") == task_name_to_update:
-                        task_id = task.get("id")
-                        print(
-                            f"Found task_id {task_id} for task name '{task_name_to_update}'"
-                        )
-                        break
-            except Exception as e:
-                print(f"Error getting task ID from name: {e}")
-
-        # 6. Hardcoded fallback for testing
-        if not task_id:
-            # If the request is specifically about TEST TASK 2, use its ID
-            if task_name_to_update == "TEST TASK 2":
-                task_id = "86a702gha"  # Known ID of TEST TASK 2
-                print(f"Using hardcoded task_id for 'TEST TASK 2': {task_id}")
-            # For general testing, use a fallback ID
-            elif task_name_to_update == "TEST TASK":
-                task_id = "86a700c6e"  # Known ID of TEST TASK
-                print(f"Using hardcoded task_id for 'TEST TASK': {task_id}")
-            # If still no task_id, attempt to get the first task from the list
-            else:
-                try:
-                    get_tasks_tool = GetTasksTool()
-                    tasks = get_tasks_tool._run(list_id=901307715461)
-                    if tasks and len(tasks) > 0:
-                        task_id = tasks[0].get("id")
-                        print(f"Using first task from list as fallback: {task_id}")
-                except Exception as e:
-                    print(f"Error getting fallback task ID: {e}")
-
-        if not task_id:
-            raise ToolException("task_id is required for updating a task")
-
-        print(f"task_id being used: {task_id}")
-        print("==== End parameters ====\n")
-
-        action = CustomUpdateTask()
-
-        url = f"{action.url}{action.path}".format(task_id=task_id)
-        print(f"URL being used: {url}")
-
-        # Add update parameters from kwargs
-        for key, value in kwargs.items():
-            if value is not None and key != "kwargs" and key != "task_id":
-                update_params[key] = ensure_serializable(value)
-
-        # Make sure all parameters are JSON serializable
-        params = {
-            k: ensure_serializable(v) for k, v in update_params.items() if v is not None
-        }
-
-        print(f"Update parameters: {params}")
-
-        response = requests.put(url, headers=self.headers, json=params)
-        print(f"Response status code: {response.status_code}")
-
-        if response.status_code == 200:
-            response_json = response.json()
-        else:
-            try:
-                response_json = response.json()
-                print(f"Error response: {response_json}")
-            except requests.JSONDecodeError:
-                response_json = {"error": "Invalid JSON response"}
-                print("Could not decode JSON response")
-
-        response = UpdateTaskResponse(data=response_json)
-        filtered_response = {
-            "id": response.data.get("id"),
-            "name": response.data.get("name"),
-            "status": response.data.get("status", {}).get("status"),
-            "assignees": response.data.get("assignees"),
-            "due_date": response.data.get("due_date"),
-            "error": response.data.get("err"),
-        }
-
-        print(f"Returning filtered response: {json.dumps(filtered_response, indent=2)}")
-
-        return filtered_response
-
-
-class AddDependencyTool(BaseTool):
-    name: str = "add_dependency_tool"
-    description: str = """
-    Tool to set a task as dependent on or blocking another task in ClickUp.
-    - Add Dependency:
-        Invoke: "AddDependencyTool" with the appropriate parameters.
-        
-    Parameters:
-    - task_id (required): The ID of the task to add dependency to
-    - depends_on (required): The ID of the task that this task depends on
-    """
-    args_schema: Type[BaseModel] = AddDependencyRequest
-    headers: dict = {"Authorization": f"{CLICKUP_TOKEN}"}
-
-    def __init__(self, **data):
-        super().__init__(**data)
-
-    def _run(self, **kwargs) -> Any:
-        """Executes adding a task dependency in ClickUp"""
-
-        # Log the received parameters to help debug
-        print("\n==== AddDependencyTool._run received parameters: ====")
-        print(f"kwargs: {kwargs}")
-
-        # Extract task_id and depends_on from different possible locations
-        task_id = None
-        depends_on = None
-        dependent_task_name = None
-        dependency_task_name = None
-
-        # 1. Direct task_id parameter
-        if "task_id" in kwargs:
-            task_id = kwargs.get("task_id")
-            print(f"Found task_id in direct parameter: {task_id}")
-
-        # 2. Direct depends_on parameter
-        if "depends_on" in kwargs:
-            depends_on = kwargs.get("depends_on")
-            print(f"Found depends_on in direct parameter: {depends_on}")
-
-        # 3. Check if parameters are inside nested kwargs
-        if "kwargs" in kwargs and isinstance(kwargs["kwargs"], dict):
-            task_id = task_id or kwargs["kwargs"].get("task_id")
-            depends_on = depends_on or kwargs["kwargs"].get("depends_on")
-            print(
-                f"Found in nested kwargs - task_id: {task_id}, depends_on: {depends_on}"
-            )
-
-        # 4. Check if there's dependency information in the kwargs object or description
-        if "kwargs" in kwargs and hasattr(kwargs["kwargs"], "description"):
-            desc = kwargs["kwargs"].description
-            # Look for dependency patterns in the description
-            dependency_match = re.search(r"'(.*?)'\s+depends\s+on\s+'(.*?)'", desc)
-            if dependency_match:
-                dependent_task_name = dependency_match.group(1).strip()
-                dependency_task_name = dependency_match.group(2).strip()
-                print(
-                    f"Found dependency in description: '{dependent_task_name}' depends on '{dependency_task_name}'"
-                )
-
-        # 5. Check any string parameters for dependency information
-        for k, v in kwargs.items():
-            if isinstance(v, str):
-                # Check if it contains direct task IDs
-                task_id_match = re.search(
-                    r'task_id[=:]\s*["\']?([0-9a-z]{8,})["\']?', v
-                )
-                if task_id_match:
-                    task_id = task_id_match.group(1)
-                    print(f"Found task_id in string parameter: {task_id}")
-
-                depends_on_match = re.search(
-                    r'depends_on[=:]\s*["\']?([0-9a-z]{8,})["\']?', v
-                )
-                if depends_on_match:
-                    depends_on = depends_on_match.group(1)
-                    print(f"Found depends_on in string parameter: {depends_on}")
-
-                # Check for task names in dependency expressions
-                dependency_match = re.search(
-                    r"['\"]?(.*?)['\"]?\s+depends\s+on\s+['\"]?(.*?)['\"]?", v
-                )
-                if dependency_match:
-                    dependent_task_name = dependency_match.group(1).strip()
-                    dependency_task_name = dependency_match.group(2).strip()
-                    print(
-                        f"Found dependency in parameter: '{dependent_task_name}' depends on '{dependency_task_name}'"
-                    )
-                    break
-
-        # 6. If we have task names but no IDs, look up the IDs
-        if (not task_id or not depends_on) and (
-            dependent_task_name or dependency_task_name
-        ):
-            try:
-                # Get all tasks in the list to find the task IDs by name
-                get_tasks_tool = GetTasksTool()
-                tasks = get_tasks_tool._run(list_id=901307715461)
-
-                # Find the dependent task by name
-                if dependent_task_name and not task_id:
-                    for task in tasks:
-                        if task.get("name") == dependent_task_name:
-                            task_id = task.get("id")
-                            print(
-                                f"Found task_id {task_id} for dependent task name '{dependent_task_name}'"
-                            )
-                            break
-
-                # Find the dependency task by name
-                if dependency_task_name and not depends_on:
-                    for task in tasks:
-                        if task.get("name") == dependency_task_name:
-                            depends_on = task.get("id")
-                            print(
-                                f"Found depends_on {depends_on} for dependency task name '{dependency_task_name}'"
-                            )
-                            break
-            except Exception as e:
-                print(f"Error getting task IDs from names: {e}")
-
-        # 7. Hardcoded fallback for testing
-        if not task_id and dependent_task_name:
-            if dependent_task_name == "TEST TASK 2":
-                task_id = "86a702gha"  # Known ID of TEST TASK 2
-                print(f"Using hardcoded task_id for 'TEST TASK 2': {task_id}")
-            elif dependent_task_name == "TEST TASK":
-                task_id = "86a700c6e"  # Known ID of TEST TASK
-                print(f"Using hardcoded task_id for 'TEST TASK': {task_id}")
-
-        if not depends_on and dependency_task_name:
-            if dependency_task_name == "TEST TASK 2":
-                depends_on = "86a702gha"  # Known ID of TEST TASK 2
-                print(f"Using hardcoded depends_on for 'TEST TASK 2': {depends_on}")
-            elif dependency_task_name == "TEST TASK":
-                depends_on = "86a700c6e"  # Known ID of TEST TASK
-                print(f"Using hardcoded depends_on for 'TEST TASK': {depends_on}")
-
-        # Check if we got both IDs we need
-        if not task_id:
-            raise ToolException("task_id is required for adding a dependency")
-
-        if not depends_on:
-            raise ToolException("depends_on is required for adding a dependency")
-
-        print(f"task_id being used: {task_id}")
-        print(f"depends_on being used: {depends_on}")
-        print("==== End parameters ====\n")
-
-        action = AddDependency()
-
-        url = f"{action.url}{action.path}".format(task_id=task_id)
-        print(f"URL being used: {url}")
-
-        # Make sure all parameters are JSON serializable
-        params = {
-            key: ensure_serializable(value)
-            for key, value in kwargs.items()
-            if value is not None and key != "kwargs" and key != "task_id"
-        }
-
-        # Create the request body with the depends_on parameter
-        request_body = {"depends_on": depends_on}
-
-        print(f"Request body: {request_body}")
-
-        response = requests.post(
-            url, headers=self.headers, params=params, json=request_body
-        )
-        print(f"Response status code: {response.status_code}")
-
-        if response.status_code == 200:
-            response_json = response.json()
-        else:
-            try:
-                response_json = response.json()
-                print(f"Error response: {response_json}")
-            except requests.JSONDecodeError:
-                response_json = {"error": "Invalid JSON response"}
-                print("Could not decode JSON response")
-
-        response = AddDependencyResponse(data=response_json)
-
-        result_message = f"Dependency added successfully: '{dependent_task_name or task_id}' depends on '{dependency_task_name or depends_on}'"
-
-        if "err" in response.data:
-            result_message = f"Error: {response.data['err']}"
-
-        print(f"Result: {result_message}")
-
-        return result_message
-
-
-class GetListTool(BaseTool):
-    name: str = "get_list_tool"
-    description: str = """
-    Tool to view information about a list in ClickUp.
-    - Get list details:
-        Invoke: "GetListTool" with the list ID as a parameter.
-        
-    Parameters:
-    - list_id (required): The ID of the list to get information about
-    """
-    args_schema: Type[BaseModel] = GetListRequest
-    headers: dict = {"Authorization": f"{CLICKUP_TOKEN}"}
-
-    def __init__(self, **data):
-        super().__init__(**data)
-
-    def _run(self, **kwargs) -> Any:
-        """Executes the request to get information about a list in ClickUp"""
-
-        # Log the received parameters to help debug
-        print("\n==== GetListTool._run received parameters: ====")
-        print(f"kwargs: {kwargs}")
-
-        # Extract list_id from different possible locations
-        list_id = None
-
-        # 1. Direct list_id parameter
-        if "list_id" in kwargs:
-            list_id = kwargs.get("list_id")
-
-        # 2. Check if list_id is inside nested kwargs
-        elif "kwargs" in kwargs and isinstance(kwargs["kwargs"], dict):
-            list_id = kwargs["kwargs"].get("list_id")
-
-        # 3. Check if list_id is in a string format in any parameter
-        for k, v in kwargs.items():
-            if isinstance(v, str) and v.isdigit():
-                try:
-                    list_id = int(v)
-                    break
-                except ValueError:
-                    pass
-
-        if not list_id:
-            raise ToolException("list_id is required for getting list information")
-
-        print(f"list_id being used: {list_id}")
-        print("==== End parameters ====\n")
-
-        action = GetList()
-
-        url = f"{action.url}{action.path}".format(list_id=list_id)
-        print(f"URL being used: {url}")
-
-        response = requests.get(url, headers=self.headers)
-        print(f"Response status code: {response.status_code}")
-
-        if response.status_code == 200:
-            response_json = response.json()
-        else:
-            try:
-                response_json = response.json()
-                print(f"Error response: {response_json}")
-            except requests.JSONDecodeError:
-                response_json = {"error": "Invalid JSON response"}
-                print("Could not decode JSON response")
-
-        response = GetListResponse(data=response_json)
-        filtered_response = {
-            "list_id": response.data.get("id"),
-            "list_name": response.data.get("name"),
-            "folder_id": response.data.get("folder", {}).get("id"),
-            "folder_name": response.data.get("folder", {}).get("name"),
-            "error": response.data.get("err"),
-        }
-        return filtered_response
-
-
-class GetTasksTool(BaseTool):
-    name: str = "get_tasks_tool"
-    description: str = """
-    Tool to view tasks in a list in ClickUp.
-    - Get tasks:
-        Invoke: "GetTasksTool" with the list ID and optional parameters.
-        
-    Parameters:
-    - list_id (required): The ID of the list to get tasks from. Example: 901307715461
-    """
-    args_schema: Type[BaseModel] = GetTasksRequest
-    headers: dict = {"Authorization": f"{CLICKUP_TOKEN}"}
-
-    def __init__(self, **data):
-        super().__init__(**data)
-
-    def _run(self, **kwargs) -> Any:
-        """Executes the request to get filtered tasks in a list in ClickUp"""
-
-        # Log the received parameters to help debug
-        print("\n==== GetTasksTool._run received parameters: ====")
-        print(f"kwargs: {kwargs}")
-
-        # Try to extract list_id from different places
-        list_id = None
-
-        # 1. Direct list_id parameter
-        if "list_id" in kwargs:
-            list_id = kwargs.get("list_id")
-
-        # 2. Check if list_id is inside nested kwargs
-        elif "kwargs" in kwargs and isinstance(kwargs["kwargs"], dict):
-            list_id = kwargs["kwargs"].get("list_id")
-
-        # 3. Check if list_id is in a string format
-        for k, v in kwargs.items():
-            if isinstance(v, str) and "901307715461" in v:
-                list_id = "901307715461"
-                break
-
-        # 4. Hardcoded fallback for this specific test case
-        if not list_id:
-            print("No list_id found in parameters, using hardcoded value 901307715461")
-            list_id = 901307715461
-
-        print(f"list_id being used: {list_id}")
-        print("==== End parameters ====\n")
-
-        action = GetTasks()
-
-        url = f"{action.url}{action.path}".format(list_id=list_id)
-
-        # Log the constructed URL
-        print(f"URL being used: {url}")
-
-        # Make sure all parameters are JSON serializable
-        query_params = {
-            k: ensure_serializable(v)
-            for k, v in kwargs.items()
-            if v is not None and k != "kwargs"
-        }
-
-        response = requests.get(url, headers=self.headers, params=query_params)
-
-        # Log the response status code
-        print(f"Response status code: {response.status_code}")
-
-        if response.status_code == 200:
-            response_json = response.json()
-        else:
-            try:
-                response_json = response.json()
-                # Log error response
-                print(f"Error response: {response_json}")
-            except requests.JSONDecodeError:
-                response_json = {"error": "Invalid JSON response"}
-                print("Could not decode JSON response")
-
-        response = GetTasksResponse(data=response_json)
-        filtered_response = []
-        for task in response.data.get("tasks", []):
-            task_info = {
-                "id": task.get("id"),
-                "name": task.get("name"),
-                "assignees": [
-                    assignee.get("username") for assignee in task.get("assignees", [])
-                ],
-                "due_date": task.get("due_date"),
-                "date_created": task.get("date_created"),
-                "status": task.get("status", {}).get("status"),
-                "url": f"https://app.clickup.com/t/{task.get('id')}",
-            }
-            filtered_response.append(task_info)
-
-        # Add error information if present
-        if response.data.get("error"):
-            filtered_response.append({"error": response.data.get("error")})
-
-        # Log the final result we're returning
-        print(f"Returning filtered response with {len(filtered_response)} items")
-
-        return filtered_response
-
-
-# Util for converting dates
-@tool
-def date_to_timestamp(date_str: str) -> int:
-    """
-    ALWAYS use this tool to convert dates from 'YYYY-MM-DD' to Unix millisecond timestamps when setting dates on ClickUp
-
-    :param date_str: Date in the format YYYY-MM-DD
-    :return: Unix timestamp in milliseconds
-    """
-    # Convert the date string to a datetime object
-    date = datetime.datetime.strptime(date_str, "%Y-%m-%d")
-
-    # Get the timestamp in seconds and convert to milliseconds
-    timestamp_ms = int(date.timestamp() * 1000)
-
-    return timestamp_ms
-
-
-def initialize_clickup_tools():
-    clickup_tools = [
-        CreateTaskTool(),
-        DeleteTaskTool(),
-        UpdateTaskTool(),
-        AddDependencyTool(),
-        GetListTool(),
-        GetTasksTool(),
-        GetTaskTool(),
-        date_to_timestamp,
-    ]
-    agency_swarm_clickup_tools = convert_langchain_tools(clickup_tools)
-    return agency_swarm_clickup_tools

agency_ai_demo/agents/ClickUpAgent/tools/ExampleTool.py

@@ -1,30 +0,0 @@
-from agency_swarm.tools import BaseTool
-from pydantic import Field
-import os
-
-account_id = "MY_ACCOUNT_ID"
-api_key = os.getenv("MY_API_KEY") # or access_token = os.getenv("MY_ACCESS_TOKEN")
-
-class ExampleTool(BaseTool):
-    """
-    A brief description of what the custom tool does.
-    The docstring should clearly explain the tool's purpose and functionality.
-    It will be used by the agent to determine when to use this tool.
-    """
-
-    # Define the fields with descriptions using Pydantic Field
-    example_field: str = Field(
-        ..., description="Description of the example field, explaining its purpose and usage for the Agent."
-    )
-
-    def run(self):
-        """
-        The implementation of the run method, where the tool's main functionality is executed.
-        This method should utilize the fields defined above to perform the task.
-        Docstring is not required for this method and will not be used by the agent.
-        """
-        # Your custom tool logic goes here
-        # do_something(self.example_field, api_key, account_id)
-
-        # Return the result of the tool's operation as a string
-        return "Result of ExampleTool operation"

agency_ai_demo/agents/ClickUpAgent/tools/GetTaskTool.py

@@ -1,78 +0,0 @@
-import os
-import sys
-import requests
-from dotenv import load_dotenv
-from langchain.tools import BaseTool
-from langchain_core.pydantic_v1 import BaseModel
-from langchain_core.tools import ToolException
-from typing import Any, Type
-import re
-from composio.tools.local.clickup.actions.get_task import (
-    GetTask,
-    GetTaskRequest,
-    GetTaskResponse,
-)
-
-# Add the parent directory to sys.path to enable imports
-sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
-# Add project utils
-from utils.tool_wrapper import convert_langchain_tools
-from utils.ensure_serializable import ensure_serializable
-
-load_dotenv()
-
-
-CLICKUP_TOKEN = os.getenv("CLICKUP_TOKEN")
-
-
-# TODO: FIX THIS TOOL
-class GetTaskTool(BaseTool):
-    name: str = "get_task_tool"
-    description: str = """
-    Ferramenta para visualizar detalhes de uma tarefa no ClickUp.
-    - Obter detalhes da tarefa:
-        Invocar: "GetTaskTool" com o ID da tarefa e parâmetros opcionais.
-    """
-    args_schema: Type[BaseModel] = GetTaskRequest
-    headers: dict = {"Authorization": f"{CLICKUP_TOKEN}"}
-
-    def _init_(self, **data):
-        super()._init_(**data)
-
-    def _run(self, **kwargs) -> Any:
-        """Executa a requisição para obter detalhes de uma tarefa no ClickUp"""
-
-        action = GetTask()
-
-        url = f"{action.url}{action.path}".format(task_id=kwargs.get("task_id"))
-
-        query_params = {k: v for k, v in kwargs.items() if v is not None}
-
-        response = requests.get(url, headers=self.headers, params=query_params)
-
-        if response.status_code == 200:
-            response_json = response.json()
-        else:
-            try:
-                response_json = response.json()
-            except requests.JSONDecodeError:
-                response_json = {"error": "Invalid JSON response"}
-
-        response = GetTaskResponse(data=response_json)
-
-        filtered_response = {
-            "id": response.data.get("id"),
-            "name": response.data.get("name"),
-            "status": response.data.get("status", {}).get("status"),
-            "due_date": response.data.get("due_date"),
-            "date_created": response.data.get("date_created"),
-            "url": response.data.get("url"),
-            "assignees": [
-                {"id": assignee.get("id"), "username": assignee.get("username")}
-                for assignee in response.data.get("assignees", [])
-            ],
-            "error": response.data.get("error"),
-        }
-
-        return filtered_response

agency_ai_demo/agents/{ClickUpAgent/ClickUpAgent.py → NotionProjectAgent/NotionProjectAgent.py}

@@ -1,18 +1,36 @@
-from agency_swarm.agents import Agent
+import os
 
+from agency_swarm.agents import Agent
+from .tools.GetTasks import GetTasksTool
 
-class ClickUpAgent(Agent):
+class NotionProjectAgent(Agent):
     def __init__(self):
+        # Retrieve the Notion integration secret from the environment
+        try:
+            integration_secret = os.environ["NOTION_INTEGRATION_SECRET"]
+        except KeyError:
+            raise EnvironmentError(
+                "NOTION_INTEGRATION_SECRET environment variable is not set."
+            )
+
+        # Initialize the parent Agent class with updated parameters
         super().__init__(
-            name="ClickUpAgent",
-            description="Project Management Assistant who interacts with ClickUp for managing project tasks",
+            name="NotionProjectAgent",
+            description="Project Management Assistant who tracks and updates project progress on Notion",
             instructions="./instructions.md",
             files_folder="./files",
             schemas_folder="./schemas",
-            tools=[],
+            tools=[GetTasksTool],
             tools_folder="./tools",
+            model="gpt-4o",
             temperature=0.3,
             max_prompt_tokens=25000,
+            # api_headers={
+            #     "notion_tasks.json": {
+            #         "Authorization": f"Bearer {integration_secret}",
+            #         "Notion-Version": "2022-06-28",
+            #     }
+            # },
         )
 
     def response_validator(self, message):

agency_ai_demo/agents/NotionProjectAgent/__init__.py

@@ -0,0 +1 @@
+from .NotionProjectAgent import NotionProjectAgent

agency_ai_demo/agents/NotionProjectAgent/instructions.md

@@ -0,0 +1,30 @@
+# NotionProjectAgent Instructions
+
+Your team uses Notion to manage projects and tasks.
+Users often refer to tasks by name, and sometimes there may not be an exact match, so you should look for the closest ones. If in doubt, provide the user with valid options to choose from or ask for more information if necessary.
+
+## NOTION_STRUCTURE
+
+-> Database: The highest level of organization in Notion. Contains all your tasks.
+--> Page: A task in a Notion database, assignable with due dates.
+--> Subpage: A child page of a parent Page, assignable to different people.
+
+## DEFAULT_NOTION_IDS
+
+Use these IDs unless specified otherwise
+
+- Database ID: 1a88235ee2ff801e8f93d8ab2e14de1d
+
+## DATABASE PROPERTIES
+
+- Task Name
+- Status
+- Priority
+- Due Date
+- Assigned To
+
+## WORKFLOWS
+
+### Create a high level WBS
+
+When required to create a WBS, you may be prompted with information about the project scope and requirements and/or you may provided with information in a task (page) in the database. Understand the project and create a high level WBS containing 5 to 10 tasks, which cover the project scope for start to end.

agency_ai_demo/agents/NotionProjectAgent/tools/GetTasks.py

@@ -0,0 +1,136 @@
+import os
+from dotenv import load_dotenv
+from agency_swarm.tools import BaseTool
+from pydantic import Field
+
+load_dotenv()
+
+notion_integration_secret = os.getenv("NOTION_INTEGRATION_SECRET")
+notion_database_id = os.getenv("NOTION_DATABASE_ID")
+
+
+class GetTasksTool(BaseTool):
+    """
+    Tool for retrieving tasks from a Notion database.
+    This tool allows querying tasks with optional filtering and sorting capabilities
+    based on properties like Due Date, Status, Priority, and Task Name.
+    """
+
+    # Add example_field with a default value to satisfy BaseTool validation
+    example_field: str = Field(
+        default="notion_tasks",
+        description="Identifier for this tool. Can be left at its default value."
+    )
+
+    status: str = Field(
+        default=None,
+        description="Filter tasks by status. Options: Backlog, In Progress, In Review, Testing, Completed."
+    )
+    
+    priority: str = Field(
+        default=None,
+        description="Filter tasks by priority. Options: High, Medium, Low."
+    )
+    
+    due_date_before: str = Field(
+        default=None,
+        description="Filter tasks due before this date (format: YYYY-MM-DD)."
+    )
+    
+    due_date_after: str = Field(
+        default=None,
+        description="Filter tasks due after this date (format: YYYY-MM-DD)."
+    )
+    
+    sort_by: str = Field(
+        default="Due Date",
+        description="Property to sort by. Options: Due Date, Status, Priority, Task Name."
+    )
+    
+    sort_direction: str = Field(
+        default="ascending",
+        description="Sort direction. Options: ascending, descending."
+    )
+
+    def run(self):
+        """
+        Query a Notion database for tasks with optional filtering and sorting.
+
+        Returns:
+            dict: The JSON response from the Notion API containing the tasks.
+        """
+        import requests
+        
+        # Use the database ID from the environment variable
+        database_id = os.getenv("NOTION_DATABASE_ID")
+
+        # Set up the API endpoint
+        url = f"https://api.notion.com/v1/databases/{database_id}/query"
+
+        # Set up the headers
+        headers = {
+            "Authorization": f"Bearer {os.getenv('NOTION_INTEGRATION_SECRET')}",
+            "Notion-Version": "2022-06-28",
+            "Content-Type": "application/json",
+        }
+
+        # Prepare the request body
+        data = {}
+        
+        # Build filter
+        filters = []
+        
+        if self.status:
+            filters.append({
+                "property": "Status",
+                "status": {
+                    "equals": self.status
+                }
+            })
+            
+        if self.priority:
+            filters.append({
+                "property": "Priority",
+                "select": {
+                    "equals": self.priority
+                }
+            })
+            
+        if self.due_date_before:
+            filters.append({
+                "property": "Due Date",
+                "date": {
+                    "before": self.due_date_before
+                }
+            })
+            
+        if self.due_date_after:
+            filters.append({
+                "property": "Due Date",
+                "date": {
+                    "after": self.due_date_after
+                }
+            })
+        
+        if filters:
+            if len(filters) > 1:
+                data["filter"] = {
+                    "and": filters
+                }
+            else:
+                data["filter"] = filters[0]
+        
+        # Add sorting
+        if self.sort_by:
+            data["sorts"] = [
+                {
+                    "property": self.sort_by,
+                    "direction": self.sort_direction
+                }
+            ]
+
+        # Make the request
+        response = requests.post(url, headers=headers, json=data)
+
+        # Return the JSON response
+        return response.json()

agency_ai_demo/agents/ResearchAndReportAgent/ResearchAndReportAgent.py

@@ -11,6 +11,7 @@ class ResearchAndReportAgent(Agent):
             schemas_folder="./schemas",
             tools=[],
             tools_folder="./tools",
+            model="gpt-4o",
             temperature=0.3,
             max_prompt_tokens=25000,
         )

agency_ai_demo/agents/TechnicalProjectManager/TechnicalProjectManager.py

@@ -11,6 +11,7 @@ class TechnicalProjectManager(Agent):
             schemas_folder="./schemas",
             tools=[],
             tools_folder="./tools",
+            model="gpt-4o",
             temperature=0.3,
             max_prompt_tokens=25000,
         )

agency_ai_demo/shared_files/agency_manifesto.md

@@ -0,0 +1,122 @@
+# VRSEN AI: Company Profile
+
+## Company Background and Mission
+
+VRSEN AI is a boutique artificial intelligence agency founded by Arsenii Shatokhin. It began as a freelance venture and quickly grew into a dedicated agency with a team of around seven professionals. The company is headquartered in Tbilisi, Georgia, and positions itself as a partner for businesses looking to navigate the AI revolution.
+
+VRSEN AI's mission is straightforward: to help businesses make the most of AI by providing personalized tools and solutions that fit each client's unique needs. They strive to be the top choice for businesses seeking custom AI solutions, leveraging state-of-the-art models (especially OpenAI's technology) to drive growth and efficiency. In line with this mission, VRSEN emphasizes a client-centric approach – taking time to understand each business and making AI adoption "easy" through tailored solutions.
+
+## Products and Solutions
+
+VRSEN AI offers bespoke AI solutions that often center around generative AI and AI agents. The company's services are packaged into several solution offerings, including:
+
+- **AI Support Hub**: A custom AI-powered support system that can instantly answer customer questions and reduce support ticket volume, helping cut customer support costs. This solution is aimed at improving customer service and reducing churn by providing quick, AI-driven support responses.
+
+- **ChatGPT Website Widget**: An integrated chatbot widget for websites that uses ChatGPT to engage visitors. It personalizes user interactions and can boost sales by guiding customers or answering their queries in real time.
+
+- **ChatGPT for Business**: A tailored deployment of ChatGPT integrated with a company's internal databases and knowledge. This allows organizations to have their own branded AI assistant that can leverage proprietary data for tasks like answering employee questions or assisting with analytics.
+
+- **AI Consultation**: Advisory services where VRSEN's experts help clients clarify project requirements and craft a straightforward plan to bring an AI idea to life. This is essentially a consulting engagement to roadmap AI adoption or development strategies.
+
+All these offerings are powered by advanced AI models (notably OpenAI's GPT technology) and are customized per client. VRSEN AI highlights that dozens of clients have implemented these solutions (e.g. 52 clients for the Support Hub, 35 for the Website Widget, etc., as indicated on their site).
+
+In addition to bespoke solutions, VRSEN is developing a software platform for AI agents. The company refers to this SaaS product as "Agency AI," described as the first platform for deploying reliable AI-driven agencies at scale. Agency AI is built on VRSEN's real-world experience delivering production-ready AI agents, and it aims to let businesses spin up "agents-as-a-service" quickly and efficiently. This platform leverages VRSEN's internal frameworks (such as the open-source Agency Swarm framework) to orchestrate multiple AI agents working collaboratively. By investing in this product, VRSEN is moving beyond one-off projects into a scalable service model, which is a recent strategic development for the company.
+
+## Key Clients and Success Stories
+
+VRSEN AI's client base ranges from startups to established enterprises, and the company showcases several high-profile partners and success stories. Notably, VRSEN highlights collaborations with companies like Stripe, Cisco, Product Marketing Alliance, ESM, and Hugo Pfohe. These partnerships suggest that even large tech firms (e.g. Cisco and Stripe) and industry communities (Product Marketing Alliance) have trusted VRSEN for AI solutions.
+
+The company's case studies provide insight into the value delivered:
+
+- **ESM (Enterprise Strategic Management)**: VRSEN developed a GPT-powered querying tool for ESM's vast demographic datasets. This "GPT-Query" solution transformed data analysis for the client – what used to take hours was reduced to about 5 minutes using AI. The integration of machine learning at ESM was significantly streamlined thanks to VRSEN's partnership.
+
+- **Product Marketing Alliance (PMA)**: VRSEN built a tailored web application with AI Q&A capabilities for PMA's online community. This allowed PMA's members to retrieve information from the community's content archives instantly, instead of spending hours on manual searches. The result was a much more efficient information retrieval process for users.
+
+- **Hugo Pfohe (Automotive Dealership)**: VRSEN implemented an AI-powered chatbot for an e-commerce platform in the automotive domain (likely for Hugo Pfohe, a car dealership). The chatbot was integrated with a live SQL database of vehicles to offer personalized car recommendations. This drove improvements in sales, user engagement, and lead generation by providing instant, tailored responses to potential car buyers.
+
+Client testimonials on VRSEN's site reinforce these successes. For example, the VP of Product Marketing at one client attested that Arsenii and his team's AI solution led to "significant growth and efficiency" in their business. Another client CEO noted VRSEN's hands-on approach and valuable guidance throughout the project. These stories and endorsements illustrate VRSEN's track record in delivering impactful AI solutions across different industries.
+
+## Position in the AI Industry
+
+Within the AI industry, VRSEN AI occupies a niche at the intersection of AI consulting and AI product development. As a small, specialized agency, it has carved out a reputation for expertise in AI agents and large language model (LLM) applications. VRSEN explicitly markets itself as being "at the forefront of the AI revolution," providing cutting-edge AI agent solutions to help businesses thrive.
+
+One of VRSEN's distinguishing features is its leadership in the emerging field of AI agents (autonomous or semi-autonomous AI programs that can perform tasks or collaborate). The company's founder, Arsenii Shatokhin, is not only an AI consultant but also the creator of Agency Swarm, an open-source agent orchestration framework. Agency Swarm was born from Arsenii's effort to automate his own AI agency's workflows through AI. It allows developers to create a collaborative "swarm" of AI agents with distinct roles (like a virtual team consisting of roles such as CEO agent, developer agent, assistant agent, etc.) working together toward goals.
+
+This framework has gained significant popularity in the AI developer community (garnering thousands of stars on GitHub and widespread use), to the point where VRSEN notes that Arsenii is "leading one of the most popular AI Agent Development Frameworks called Agency Swarm."
+
+VRSEN's active role in developing such frameworks and sharing knowledge (e.g., Arsenii's YouTube channel with AI agent tutorials and the "Generative AI Podcast" he hosts) has bolstered its position as a thought leader in AI agents. In an industry where many companies are experimenting with generative AI, VRSEN stands out by focusing on reliable, production-ready AI systems and agent-based solutions. The development of its Agency AI platform further cements its innovative stance, as this product is pioneering the deployment of multi-agent AI systems at scale for businesses.
+
+In summary, VRSEN AI's position in the industry is that of a nimble innovator – a company small in size but big in specialized expertise – giving it credibility among both clients and the AI developer community.
+
+## Competitive Advantage
+
+VRSEN AI's competitive advantage lies in its combination of deep technical expertise and a highly customized, client-focused approach. A few key differentiators include:
+
+- **Bespoke AI Solutions**: Unlike one-size-fits-all software vendors, VRSEN delivers custom AI development tailored to each client's needs. The team makes a point of closely understanding the client's business challenges and then crafting AI tools specifically for them. This personalization ensures that the solutions align with the client's workflows and goals, often integrating with the client's own data systems (as seen in their ChatGPT for Business offerings).
+
+- **Expertise in AI Agents and LLMs**: VRSEN has specialized know-how in building AI agents and leveraging large language models. Their familiarity with frameworks like OpenAI's API, LangChain, vector databases, and their own Agency Swarm gives them an edge in developing sophisticated AI applications. Few agencies of their size have in-house frameworks with community recognition; this indicates a level of technical maturity and thought leadership.
+
+- **Agility of a Small Team**: With a core team (under 10 people) of AI experts, software engineers, and UX designers, VRSEN can be very agile. They iterate quickly and often work hands-on with clients (the founder himself often engages in coding and problem-solving, according to client feedback). This nimbleness allows them to implement cutting-edge AI features faster than larger consultancies might.
+
+- **Proven Track Record with Notable Clients**: Despite its size, VRSEN has delivered results for well-known companies (from tech firms to international businesses). These case studies and testimonials serve as social proof of the agency's capability. It demonstrates that VRSEN can compete with bigger players when it comes to delivering real business value through AI.
+
+- **Education and Thought Leadership**: VRSEN (and Arsenii Shatokhin personally) actively educates the market via YouTube videos, podcasts, and open-source contributions. This not only markets their expertise but keeps them at the leading edge of AI trends. Clients benefit from this knowledge, as VRSEN's solutions are informed by the latest research and best practices in AI development.
+
+In essence, VRSEN's advantage is being both highly innovative and client-friendly. They lower the barrier for businesses to adopt AI by handling the complexity (making it "easy" for clients), all while pushing the envelope in areas like autonomous agents. This combination of service and innovation distinguishes them in the competitive AI services industry.
+
+## Corporate Culture and Values
+
+Despite its high-tech focus, VRSEN AI emphasizes a human-centric and collaborative corporate culture. The team is described as a group of "passionate innovators and problem solvers" who thrive on tackling challenging problems in AI. The work environment is collaborative and supportive, where every team member's contribution is recognized and valued. This suggests a flat or egalitarian culture typical of startups, in which open communication and teamwork are encouraged.
+
+VRSEN also values efficiency, clear communication, and continual feedback in its operations. Given that the company works with clients globally, the team is remote-friendly and flexible – team members often need to be available across various time zones for meetings or calls. This global remote setup (team members are in different countries) requires trust and strong communication skills internally. The company's job postings explicitly mention the importance of conveying complex technical ideas in simple terms for clients, reflecting a culture that bridges technical excellence with client understanding.
+
+Innovation is at the heart of VRSEN's identity, so it's likely that the culture encourages continuous learning and experimentation with new AI technologies. Being a small company, there is also an entrepreneurial spirit – employees often wear multiple hats and have significant ownership of their projects. Overall, VRSEN AI's corporate culture can be summed up as innovative, collaborative, and customer-focused: they are excited about pushing AI boundaries, work closely as a team, and remain aligned with what clients need.
+
+## Technical AI Project Manager Role at VRSEN AI
+
+At VRSEN, a Technical AI Project Manager plays a critical role in ensuring the success of AI projects from conception to delivery. This role is a blend of project leadership, technical oversight, and client management. Based on VRSEN's job description for a Technical Project Manager, the responsibilities and expectations include:
+
+- **Client Liaison**: Serving as the primary point of contact between clients and VRSEN's development team. The project manager needs to manage client communications proactively, understand client requirements, and keep the clients updated. Essentially, they "bridge the gap" between the esteemed clients and the developers.
+
+- **Project Scoping and Planning**: Working closely with clients to scope out AI projects – defining the tasks, deliverables, and timelines for AI agent development engagements. This involves translating business needs into technical requirements and creating clear project plans that the team can execute.
+
+- **Project Management and Delivery**: Overseeing the execution of projects to ensure they are delivered on time and meet quality standards. The Technical PM coordinates the development sprints, monitors progress, mitigates risks, and ensures that the AI solutions being built align with client expectations. They are responsible for timely delivery and maintaining high quality in outcomes.
+
+- **Technical Guidance and Oversight**: Although this is a management role, it has a strong technical component. The Technical AI Project Manager is expected to engage in technical discussions with clients and provide expert opinions on AI agents and related technologies. They should be comfortable reviewing the team's work, giving constructive feedback to developers, and possibly even diving into code or technical problem-solving when needed. In fact, proficiency in Python (especially backend development) and experience with AI frameworks (OpenAI API, LangChain, Hugging Face, vector databases, etc.) are listed as necessary skills for this role.
+
+- **Team Leadership and Feedback**: Providing leadership to the development team by setting expectations, clarifying tasks, and offering feedback. The project manager guides the team through challenges and ensures everyone is aligned on the project goals. They act not as a distant overseer but as a supportive leader who can mentor developers and help troubleshoot issues. Strong organizational skills are needed to juggle multiple projects simultaneously and keep everything on track.
+
+- **Flexibility and Availability**: Since VRSEN works with a global client base, the Technical Project Manager should be available for meetings or sales calls across various time zones as required. This implies some non-traditional hours at times and a flexibility in schedule to accommodate international clients.
+
+### Expected Qualifications
+
+A Technical AI Project Manager at VRSEN is generally expected to have a background that spans both engineering and project management. The job posting suggests candidates should have around 3-5 years of experience in relevant fields.
+
+Key technical qualifications include:
+
+- Strong Python skills
+- Firsthand experience building AI applications or agents with modern tools (for example, being familiar with how to use OpenAI's APIs or implement AI agents in frameworks like Agency Swarm or LangChain)
+- Knowledge of backend infrastructure (possibly some DevOps exposure) is also valuable, given the emphasis on reliable deployment of AI solutions
+
+On the soft skills side:
+
+- Exceptional communication is a must – the ability to clearly explain complex AI concepts to non-technical stakeholders
+- Being highly organized and capable of managing multiple parallel projects is also highlighted
+
+Because this role deals with cutting-edge AI projects, a passion for AI and staying updated on new developments would be expected too (even if not explicitly stated, it aligns with VRSEN's innovative culture).
+
+In terms of role within the company, the Technical AI Project Manager is pivotal for VRSEN's operations. They ensure that the client's vision is accurately translated for the technical team and that the final AI solution meets the client's business objectives. Essentially, they wear the hats of project planner, tech lead, and client advisor all at once. VRSEN's job description notes that this role is key to "bridging the gap" between clients and developers, highlighting how central the position is in delivering successful projects. Given VRSEN's collaborative ethos, a person in this role would work closely with the founder and other team leads, and they would contribute to refining VRSEN's project delivery processes over time.
+
+Overall, a Technical AI Project Manager at VRSEN AI is expected to drive projects to success by blending technical know-how with project leadership and excellent people skills. They play a major part in upholding VRSEN's reputation for quality and innovation in bespoke AI solutions, while also fostering a positive experience for clients through professional communication and reliable delivery.
+
+## Sources
+
+- VRSEN AI – Official Website (About Us) [VRSEN.AI](https://vrsen.ai)
+- VRSEN AI – Official Website (Solutions) [VRSEN.AI](https://vrsen.ai)
+- VRSEN AI – Official Website (Case Studies) [VRSEN.AI](https://vrsen.ai)
+- VRSEN AI – Official Website (Home page/client testimonials) [VRSEN.AI](https://vrsen.ai)
+- VRSEN AI – Agents-as-a-Service Careers (Technical Project Manager role) [AGENTS.VRSEN.AI](https://agents.vrsen.ai)
+- VRSEN AI – Agents-as-a-Service Careers (Technical AI Product Manager role, describing Agency AI) [AGENTS.VRSEN.AI](https://agents.vrsen.ai)
+- Agency Swarm – Open Source Framework by VRSEN (Project Introduction) [VRSEN.GITHUB.IO](https://vrsen.github.io)
+- Search result snippet referencing VRSEN founder and Agency Swarm [AGENTS.VRSEN.AI](https://agents.vrsen.ai) (confirmation of Agency Swarm's popularity and founder's leadership)
+- Client testimonial from VRSEN site (ESM partnership) [VRSEN.AI](https://vrsen.ai)

agency_ai_demo/utils/ensure_serializable.py

@@ -1,30 +0,0 @@
-import json
-
-
-# Add a utility function to ensure all values are JSON serializable
-def ensure_serializable(obj):
-    """
-    Recursively ensure that an object and all its contents are JSON serializable.
-    Handles Pydantic Field objects by extracting their values.
-    """
-    # Handle None
-    if obj is None:
-        return None
-
-    # Check if it's a Field object (has certain common Field attributes)
-    if hasattr(obj, "default") and hasattr(obj, "description"):
-        # Return the default value or None
-        if obj.default is not None:
-            return obj.default
-        return None
-
-    # Handle dictionaries
-    if isinstance(obj, dict):
-        return {k: ensure_serializable(v) for k, v in obj.items()}
-
-    # Handle lists
-    if isinstance(obj, list):
-        return [ensure_serializable(item) for item in obj]
-
-    # Return other objects as is
-    return obj

agency_ai_demo/utils/tool_wrapper.py

@@ -1,489 +0,0 @@
-import inspect
-from typing import Any, Callable, Dict, List, Optional, Type, Union, get_type_hints
-from pydantic import BaseModel, Field, create_model
-import functools
-import logging
-
-# Set up logging
-logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
-logger = logging.getLogger(__name__)
-
-# Import the necessary classes
-try:
-    # For LangChain
-    from langchain.tools import BaseTool as LangChainBaseTool
-    from langchain.tools.base import ToolException
-
-    LANGCHAIN_AVAILABLE = True
-except ImportError:
-    logger.warning("LangChain not available. Providing mock classes for documentation.")
-    LANGCHAIN_AVAILABLE = False
-
-    class LangChainBaseTool:
-        """Mock LangChain BaseTool for documentation"""
-
-        name = "mock_tool"
-        description = "Mock tool for documentation"
-
-        def _run(self, *args, **kwargs):
-            raise NotImplementedError("This is a mock class")
-
-    class ToolException(Exception):
-        """Mock ToolException for documentation"""
-
-        pass
-
-
-try:
-    # For agency-swarm
-    from agency_swarm.tools.BaseTool import BaseTool as AgencySwarmBaseTool
-
-    AGENCY_SWARM_AVAILABLE = True
-except ImportError:
-    logger.warning(
-        "Agency Swarm not available. Providing mock classes for documentation."
-    )
-    AGENCY_SWARM_AVAILABLE = False
-
-    class AgencySwarmBaseTool:
-        """Mock Agency Swarm BaseTool for documentation"""
-
-        class ToolConfig:
-            strict = True
-            one_call_at_a_time = True
-            output_as_result = False
-            async_mode = None
-
-        def run(self):
-            raise NotImplementedError("This is a mock class")
-
-
-def convert_langchain_tool(
-    lc_tool: Union[Type[LangChainBaseTool], Callable]
-) -> Type[AgencySwarmBaseTool]:
-    """
-    Converts a LangChain tool (class-based or function-based) to an agency-swarm compatible tool.
-
-    Args:
-        lc_tool: Either a LangChain BaseTool class or a function decorated with @tool
-
-    Returns:
-        A new class that inherits from agency_swarm.tools.BaseTool
-
-    Example:
-        ```python
-        from langchain.tools import BaseTool
-
-        class MyLangChainTool(BaseTool):
-            name = "my_tool"
-            description = "A sample tool"
-
-            def _run(self, param1: str, param2: int) -> str:
-                return f"Processed {param1} and {param2}"
-
-        # Convert to agency-swarm tool
-        MyAgencySwarmTool = convert_langchain_tool(MyLangChainTool)
-
-        # Use with agency-swarm
-        agent = Agent(
-            name="my_agent",
-            tools=[MyAgencySwarmTool],  # Pass the class, not an instance
-        )
-        ```
-    """
-    if not LANGCHAIN_AVAILABLE or not AGENCY_SWARM_AVAILABLE:
-        raise ImportError(
-            "Both LangChain and Agency Swarm must be available to convert tools."
-        )
-
-    # Check if input is a function (likely decorated with @tool)
-    if callable(lc_tool) and not inspect.isclass(lc_tool):
-        return _convert_function_tool(lc_tool)
-
-    # If it's a class, ensure it's a subclass of LangChain's BaseTool
-    if not issubclass(lc_tool, LangChainBaseTool):
-        raise TypeError(f"Expected a LangChain BaseTool subclass, got {lc_tool}")
-
-    # Extract metadata from the LangChain tool
-    tool_name = getattr(lc_tool, "name", lc_tool.__name__)
-    if tool_name is None:
-        tool_name = lc_tool.__name__
-
-    tool_description = getattr(lc_tool, "description", lc_tool.__doc__ or "")
-
-    # Get the schema from the LangChain tool, if any
-    schema_cls = getattr(lc_tool, "args_schema", None)
-
-    # Create a new class that inherits from agency_swarm's BaseTool
-    class ConvertedTool(AgencySwarmBaseTool):
-        """
-        Agency Swarm tool converted from LangChain tool.
-        """
-
-        # Set up the ToolConfig inner class
-        class ToolConfig:
-            strict: bool = True
-            one_call_at_a_time: bool = True
-            output_as_result: bool = False
-            async_mode: None = None
-
-        def run(self) -> Dict:
-            """
-            Execute the tool with runtime validation.
-            """
-            # Validate required fields based on the original schema
-            validation_errors = self._validate_required_fields()
-            if validation_errors:
-                return {"success": False, "error": validation_errors}
-
-            # Prepare args for the original tool's _run method
-            kwargs = {
-                field: getattr(self, field)
-                for field in self._get_field_names()
-                if hasattr(self, field) and getattr(self, field) is not None
-            }
-
-            try:
-                # Call the original LangChain tool's _run method
-                instance = lc_tool()
-                result = instance._run(**kwargs)
-
-                # If result is already a dict, return it
-                if isinstance(result, dict):
-                    return result
-
-                # Otherwise, wrap it in a success response
-                return {"success": True, "result": result}
-            except ToolException as e:
-                # Convert LangChain's exceptions to structured errors
-                return {"success": False, "error": str(e)}
-            except Exception as e:
-                # Handle unexpected errors
-                logger.exception(f"Tool execution failed: {str(e)}")
-                return {"success": False, "error": f"Tool execution failed: {str(e)}"}
-
-        def _validate_required_fields(self) -> Optional[str]:
-            """
-            Validate required fields at runtime.
-            """
-            if not schema_cls:
-                return None
-
-            # Try different approaches to identify required fields based on Pydantic version
-            missing_fields = []
-            try:
-                # Attempt to get field info based on Pydantic v1 style
-                if hasattr(schema_cls, "__fields__"):
-                    for field_name, field_info in schema_cls.__fields__.items():
-                        if (
-                            field_info.required
-                            and getattr(self, field_name, None) is None
-                        ):
-                            missing_fields.append(field_name)
-                # Try Pydantic v2 style
-                elif hasattr(schema_cls, "model_fields"):
-                    for field_name, field_info in schema_cls.model_fields.items():
-                        if (
-                            field_info.is_required()
-                            and getattr(self, field_name, None) is None
-                        ):
-                            missing_fields.append(field_name)
-                # Fallback to checking for ... (Ellipsis) in class variables
-                else:
-                    for field_name in schema_cls.__annotations__:
-                        class_var = getattr(schema_cls, field_name, None)
-                        if class_var is ... and getattr(self, field_name, None) is None:
-                            missing_fields.append(field_name)
-            except Exception as e:
-                logger.warning(f"Error during field validation: {e}")
-                # If all else fails, just check for None values
-                for field_name in schema_cls.__annotations__:
-                    if getattr(self, field_name, None) is None:
-                        missing_fields.append(field_name)
-
-            if missing_fields:
-                return f"Missing required fields: {', '.join(missing_fields)}"
-
-            return None
-
-        def _get_field_names(self) -> List[str]:
-            """
-            Get all field names from the schema.
-            """
-            if schema_cls:
-                return list(schema_cls.__annotations__.keys())
-            return []
-
-    # Set the tool name and description
-    # Use safe string operations for the class name
-    safe_name = "".join(
-        c for c in tool_name.replace("-", "_") if c.isalnum() or c == "_"
-    )
-    if safe_name:
-        class_name = safe_name[0].upper() + safe_name[1:] + "Converted"
-    else:
-        class_name = "ConvertedTool"
-
-    ConvertedTool.__name__ = class_name
-    ConvertedTool.__doc__ = tool_description
-
-    # Add fields to the converted tool
-    if schema_cls:
-        _add_fields_from_schema(ConvertedTool, schema_cls)
-    else:
-        # If no schema_cls, try to infer from _run signature
-        _add_fields_from_run_method(ConvertedTool, lc_tool)
-
-    return ConvertedTool
-
-
-def _add_fields_from_schema(
-    target_cls: Type[AgencySwarmBaseTool], schema_cls: Type[BaseModel]
-) -> None:
-    """
-    Add fields from a Pydantic schema to the target class.
-
-    Args:
-        target_cls: The class to add fields to
-        schema_cls: The Pydantic schema class to extract fields from
-    """
-    target_cls.__annotations__ = {}
-
-    # Extract field descriptions based on Pydantic version
-    field_descriptions = {}
-    try:
-        if hasattr(schema_cls, "__fields__"):
-            # Pydantic v1
-            for name, field in schema_cls.__fields__.items():
-                field_descriptions[name] = field.field_info.description
-        elif hasattr(schema_cls, "model_fields"):
-            # Pydantic v2
-            for name, field in schema_cls.model_fields.items():
-                field_descriptions[name] = field.description
-    except Exception as e:
-        logger.warning(f"Error extracting field descriptions: {e}")
-
-    # Add each field to the target class
-    for field_name, field_type in schema_cls.__annotations__.items():
-        description = field_descriptions.get(field_name, f"Parameter: {field_name}")
-
-        # Make the type Optional if it's not already
-        if not hasattr(field_type, "__origin__") or field_type.__origin__ is not Union:
-            field_type = Optional[field_type]
-
-        # Add the field to the target class with None as default
-        setattr(target_cls, field_name, Field(None, description=description))
-        target_cls.__annotations__[field_name] = field_type
-
-
-def _add_fields_from_run_method(
-    target_cls: Type[AgencySwarmBaseTool], tool_cls: Type[LangChainBaseTool]
-) -> None:
-    """
-    Add fields inferred from the _run method signature.
-
-    Args:
-        target_cls: The class to add fields to
-        tool_cls: The LangChain tool class to extract fields from
-    """
-    target_cls.__annotations__ = {}
-
-    # Get the signature of the _run method
-    try:
-        # First try to get the _run method from the class
-        if hasattr(tool_cls, "_run"):
-            run_method = tool_cls._run
-        # If that fails, try to get it from an instance
-        else:
-            instance = tool_cls()
-            run_method = instance._run
-
-        signature = inspect.signature(run_method)
-
-        # Skip 'self' parameter if present
-        parameters = list(signature.parameters.items())
-        if parameters and parameters[0][0] == "self":
-            parameters = parameters[1:]
-
-        for param_name, param in parameters:
-            # Skip *args and **kwargs
-            if param.kind in (
-                inspect.Parameter.VAR_POSITIONAL,
-                inspect.Parameter.VAR_KEYWORD,
-            ):
-                continue
-
-            # Get the type annotation if available
-            param_type = (
-                param.annotation if param.annotation != inspect.Parameter.empty else Any
-            )
-
-            # Make the type Optional
-            optional_type = Optional[param_type]
-
-            # Add the field to the target class
-            setattr(
-                target_cls,
-                param_name,
-                Field(None, description=f"Parameter: {param_name}"),
-            )
-            target_cls.__annotations__[param_name] = optional_type
-
-    except Exception as e:
-        logger.warning(f"Error extracting fields from _run method: {e}")
-
-
-def _convert_function_tool(tool_func: Callable) -> Type[AgencySwarmBaseTool]:
-    """
-    Convert a function-based tool (decorated with @tool) to an agency-swarm compatible tool.
-
-    Args:
-        tool_func: A function decorated with @tool
-
-    Returns:
-        A new class that inherits from agency_swarm.tools.BaseTool
-    """
-    # Extract metadata from the function
-    # Handle case where tool_func might be a StructuredTool instead of a direct function
-    if hasattr(tool_func, "name"):
-        tool_name = tool_func.name
-    else:
-        tool_name = getattr(tool_func, "__name__", "function_tool")
-
-    tool_description = getattr(
-        tool_func, "description", getattr(tool_func, "__doc__", "") or ""
-    )
-
-    # Get the signature of the underlying function
-    if hasattr(tool_func, "_run"):
-        func_to_inspect = tool_func._run
-    elif hasattr(tool_func, "func"):
-        func_to_inspect = tool_func.func
-    else:
-        func_to_inspect = tool_func
-
-    signature = inspect.signature(func_to_inspect)
-
-    # Create a new class that inherits from agency_swarm's BaseTool
-    class ConvertedFunctionTool(AgencySwarmBaseTool):
-        """
-        Agency Swarm tool converted from a LangChain function tool.
-        """
-
-        # Set up the ToolConfig inner class
-        class ToolConfig:
-            strict: bool = True
-            one_call_at_a_time: bool = True
-            output_as_result: bool = False
-            async_mode: None = None
-
-        def run(self) -> Dict:
-            """
-            Execute the tool with runtime validation.
-            """
-            # Prepare args for the original function
-            kwargs = {
-                param_name: getattr(self, param_name, None)
-                for param_name in signature.parameters
-                if hasattr(self, param_name) and getattr(self, param_name) is not None
-            }
-
-            # Validate required parameters
-            missing_params = []
-            for param_name, param in signature.parameters.items():
-                if (
-                    param.default == inspect.Parameter.empty  # No default value
-                    and kwargs.get(param_name) is None  # Not provided
-                ):
-                    missing_params.append(param_name)
-
-            if missing_params:
-                return {
-                    "success": False,
-                    "error": f"Missing required parameters: {', '.join(missing_params)}",
-                }
-
-            try:
-                # Call the original function or tool
-                if hasattr(tool_func, "_run"):
-                    result = tool_func._run(**kwargs)
-                elif hasattr(tool_func, "__call__"):
-                    result = tool_func(**kwargs)
-                else:
-                    result = func_to_inspect(**kwargs)
-
-                # If result is already a dict, return it
-                if isinstance(result, dict):
-                    return result
-
-                # Otherwise, wrap it in a success response
-                return {"success": True, "result": result}
-            except Exception as e:
-                # Handle errors
-                logger.exception(f"Tool execution failed: {str(e)}")
-                return {"success": False, "error": f"Tool execution failed: {str(e)}"}
-
-    # Set the tool name and description
-    # Use safe string operations for the class name
-    safe_name = "".join(
-        c for c in tool_name.replace("-", "_") if c.isalnum() or c == "_"
-    )
-    if safe_name:
-        class_name = safe_name[0].upper() + safe_name[1:] + "Tool"
-    else:
-        class_name = "ConvertedFunctionTool"
-
-    ConvertedFunctionTool.__name__ = class_name
-    ConvertedFunctionTool.__doc__ = tool_description
-
-    # Set up annotations dictionary
-    ConvertedFunctionTool.__annotations__ = {}
-
-    # Add fields based on function parameters
-    for param_name, param in signature.parameters.items():
-        # Get the type annotation if available
-        param_type = (
-            param.annotation if param.annotation != inspect.Parameter.empty else Any
-        )
-
-        # Make it Optional
-        optional_type = Optional[param_type]
-
-        # Get description from docstring if available
-        param_description = f"Parameter: {param_name}"
-
-        # Add the field to the class
-        setattr(
-            ConvertedFunctionTool,
-            param_name,
-            Field(None, description=param_description),
-        )
-        ConvertedFunctionTool.__annotations__[param_name] = optional_type
-
-    return ConvertedFunctionTool
-
-
-# Batch conversion helper
-def convert_langchain_tools(
-    lc_tools: List[Union[Type[LangChainBaseTool], Callable]]
-) -> List[Type[AgencySwarmBaseTool]]:
-    """
-    Convert multiple LangChain tools to agency-swarm compatible tools.
-
-    Args:
-        lc_tools: A list of LangChain BaseTool classes or functions decorated with @tool
-
-    Returns:
-        A list of converted agency-swarm tool classes
-    """
-    converted_tools = []
-    for tool in lc_tools:
-        try:
-            converted_tool = convert_langchain_tool(tool)
-            converted_tools.append(converted_tool)
-        except Exception as e:
-            logger.error(
-                f"Failed to convert tool {getattr(tool, 'name', getattr(tool, '__name__', 'unknown'))}: {e}"
-            )
-
-    return converted_tools