Spaces:
Runtime error
Runtime error
File size: 3,460 Bytes
f2389eb d2aba87 f2389eb 5531fc6 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 f2389eb d2aba87 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 |
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, 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()
|