Building a Natural Language Task Scheduler with FastAPI, Notion, and LLM

Are you tired of manually creating and managing tasks in Notion? What if you could just say "Schedule a meeting with John tomorrow at 3pm" and have it automatically create a task? In this tutorial, we'll build exactly that - a natural language task scheduler that integrates with Notion!

What We're Building

We'll create a system that can:

• Create tasks using natural language
• Delete tasks when you ask
• Reschedule tasks using simple commands
• Integrate with your Notion workspace

For example, you can say:

"Schedule team meeting with John tomorrow at 3pm"
"Delete my meeting with Sarah"
"Reschedule project review to next Monday at 2pm"

Prerequisites

Before we start, make sure you have:

• Python 3.8+ installed
• A Notion account and API key
• A Groq API key (for LLM integration)
• Basic understanding of Python and APIs

Project Setup

1. First, create your project structure:

notion-ai-scheduler/
├── app/
│   ├── services/
│   │   ├── notion_service.py
│   │   ├── task_processor.py
│   │   └── scheduler_service.py
│   ├── main.py
├── .env
└── requirements.txt

1. Install required packages:

pip install fastapi uvicorn notion-client groq python-dotenv pytz

1. Set up your environment variables (.env):

NOTION_API_KEY=your_notion_api_key
NOTION_DATABASE_ID=your_database_id
GROQ_API_KEY=your_groq_api_key

Step 1: Setting Up Notion

1. Create a new Notion database with these columns:
2. Task (Title)
3. Status (Select: Not Started, In Progress, Done)
4. Due Date (Date)

5. Priority (Select: High, Medium, Low)

6. Create a Notion integration:

7. Go to https://www.notion.so/my-integrations

8. Create a new integration

9. Copy your API key

10. Share your database with the integration

Step 2: Creating the Notion Service

Create notion_service.py:

from notion_client import Client

class NotionService:
    def __init__(self):
        self.notion = Client(auth=NOTION_API_KEY)
        self.database_id = NOTION_DATABASE_ID

    def create_task(self, title: str, due_date: str = None, priority: str = "Medium"):
        try:
            properties = {
                "Task": {"title": [{"text": {"content": title}}]},
                "Status": {"status": {"name": "Not started"}},
                "Priority": {"status": {"name": priority}}
            }
            if due_date:
                properties["Due Date"] = {"date": {"start": due_date}}

            return self.notion.pages.create(
                parent={"database_id": self.database_id},
                properties=properties
            )
        except Exception as e:
            print(f"Error creating task: {e}")
            raise

    def delete_task(self, task_id: str):
        try:
            return self.notion.pages.update(
                page_id=task_id,
                archived=True
            )
        except Exception as e:
            print(f"Error deleting task: {e}")
            raise

Step 3: Creating the LLM Processor

Create task_processor.py:

from groq import Groq
from datetime import datetime
import pytz

class TaskProcessor:
    def __init__(self):
        self.client = Groq()
        self.ist = pytz.timezone('Asia/Kolkata')

    def process_input(self, user_input: str):
        current_time = datetime.now(self.ist)

        prompt = f"""You are a task management assistant.
        Current time: {current_time.strftime("%Y-%m-%d %H:%M:%S")}

        Convert the task request into a JSON with:
        - title: task description
        - due_date: YYYY-MM-DD HH:MM format or null
        - priority: High/Medium/Low
        - needs_clarification: true if time is vague
        """

        try:
            completion = self.client.chat.completions.create(
                messages=[
                    {"role": "system", "content": prompt},
                    {"role": "user", "content": user_input}
                ],
                model="mixtral-8x7b-32768",
                response_format={"type": "json_object"}
            )

            return json.loads(completion.choices[0].message.content)
        except Exception as e:
            print(f"Error: {str(e)}")
            return None

Step 4: Creating the Scheduler Service

Create scheduler_service.py:

class SchedulerService:
    def __init__(self):
        self.notion = NotionService()
        self.processor = TaskProcessor()

    def create_task_from_text(self, text: str):
        llm_result = self.processor.process_input(text)
        if not llm_result:
            return {"error": "Failed to process text"}

        if llm_result.get('needs_clarification'):
            return {
                "needs_clarification": True,
                "question": llm_result['clarification_question']
            }

        try:
            notion_task = self.notion.create_task(
                title=llm_result['title'],
                due_date=llm_result['due_date'],
                priority=llm_result['priority']
            )
            return {
                "success": True,
                "task": notion_task,
                "message": "Task created successfully"
            }
        except Exception as e:
            return {"error": f"Failed to create task in Notion: {str(e)}"}

Step 5: Setting Up FastAPI

Create main.py:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()
scheduler = SchedulerService()

class TextInput(BaseModel):
    text: str

@app.post("/process-text")
def process_text_to_task(text_input: TextInput):
    try:
        result = scheduler.create_task_from_text(text_input.text)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Running the Project

1. Start the server:

uvicorn app.main:app --reload

1. Test creating a task:

curl -X POST "http://localhost:8000/process-text" \
-H "Content-Type: application/json" \
-d '{"text": "Schedule team meeting tomorrow at 3pm"}'

Key Features and How They Work

Natural Language Processing

• The LLM (Groq) processes natural language input
• Extracts key information like dates, times, and priorities
• Handles various date formats ("tomorrow", "next Monday", etc.)

Notion Integration

• Creates tasks in your Notion database
• Manages task status and priority
• Handles date and time conversions

Error Handling

• Validates input
• Handles API errors gracefully
• Asks for clarification when needed

Advanced Features

Task Deletion

@app.post("/process-deletion")
def process_deletion(text_input: TextInput):
    try:
        result = scheduler.task_deletion(text_input.text)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Task Rescheduling

@app.post("/reschedule-task")
def reschedule_task(text_input: TextInput):
    try:
        result = scheduler.reschedule_task(text_input.text)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Future Improvements

You could extend this project by:

1. Adding a frontend interface
2. Implementing user authentication
3. Adding support for recurring tasks
4. Creating task templates
5. Adding email notifications

Conclusion

We've built a powerful task management system that combines the simplicity of natural language with the organization of Notion. This project demonstrates how to:

• Integrate multiple APIs
• Process natural language
• Handle complex date/time operations
• Build a practical tool for daily use

The complete code is available on https://github.com/chetansingh-2/notion-ai-project.

Resources

• FastAPI Documentation
• Notion API Documentation
• Groq Documentation

Happy coding! 🚀