Building the Operations layer: a Notion and agents integration (part 1)

This post is part of a series on OPERA, a framework I am developing that puts an automation layer on top of a human-centric filing system. It comes out of my own setup, so read it as working notes rather than advice.

From theory to something running

In the previous post on OPERA I described the Operations layer as the agentic engine that coordinates tasks, curates knowledge and keeps a workflow continuous across platforms. That description is abstract enough to mean almost anything, so this post is the first concrete version of it.

It is a Notion integration that picks up tasks and processes them with specialized AI crews: a bridge between the information you already keep organized and the tools that can act on it. I will go through the whole build, including how to record the agents' intermediate reasoning so you can see how a result was reached.

It runs entirely from Notion, so there is no second tool to manage tasks in, and the only running cost is the API usage for OpenAI and Serper, because Render hosts this kind of service on a free tier.

Why Notion

The Operations layer has to connect the organized information in my knowledge base (Projects, Environments, Resources, Archives) to something that can process it automatically.

Notion is a good base for three reasons:

1. Structured data access: Notion databases store task information consistently, with properties the code can rely on.
2. API-first design: the Notion API makes reading and updating that information straightforward.
3. Familiar interface: there is no new tool to learn, since tasks are created and results read in the system you already work in.

What agents add on top of ordinary automation is judgment about which handler a task needs, and the ability to produce the content rather than just move it. Together the two give you a system that can:

• Determine which type of AI "crew" should handle a particular task
• Process tasks with appropriate AI tools and methodologies
• Update results directly back to your knowledge base

That is the Operations layer: it runs the work without disturbing how the underlying system is organized.

How the system is put together

There are four pieces:

1. A scheduled service that periodically checks Notion for tasks marked for execution
2. An orchestrator that determines which AI crew should handle each task
3. Specialized crews (like a Research Crew) that process tasks using AI agents
4. A Notion integration that updates task results back to your database

It runs as a background service on Render, whose free tier covers personal use. Every few minutes it checks the Notion database for new tasks, processes each one with the right crew, and writes the results back, with nothing for you to do in between.

The structure is modular, so you can start with one crew (the Research Crew below) and add more specialized ones later without touching what already works.

Setting up the Notion integration

The system needs an integration with permission to read and modify your workspace.

Creating the integration

1. Go to https://www.notion.so/my-integrations or navigate to Settings & Members → Integrations → Develop or install integrations
2. Click on "+ New integration"
3. Give your integration a name (e.g., "OPERA Operations Layer")
4. Select the workspace where you want to use this integration
5. Under "Capabilities", select read content, update content, insert content, read comments and create comments
6. Click "Submit" to create the integration
7. Copy the "Internal Integration Secret", which becomes your NOTION_API_KEY

Credit: Notion

Setting up the database

The database holds the tasks and is the whole interface between you and the crews.

It needs three properties:

• Status: A select property with options like "To Do", "Execute", "In Progress", "Done"
• Task: A title property containing the task description
• Response: A rich text property where results will be stored

The loop runs like this:

1. You create a task in Notion and set its status to "Execute"
2. The system detects the task during its next check
3. The system processes the task using the appropriate AI crew
4. The results are written back to Notion, and the status is updated to "Done"

Development environment

Set up the project:

# Create a new directory for this project
mkdir notion-agent-integration
cd notion-agent-integration

# Create and activate a virtual environment
# Windows
python -m venv venv
venv\Scripts\activate

# macOS/Linux
python3 -m venv venv
source venv/bin/activate

# Install required packages
pip install fastapi uvicorn notion-client openai crewai crewai-tools python-dotenv

You'll need several API keys for this project:

1. Notion API Key: For accessing your Notion workspace
2. OpenAI API Key: For AI processing
3. Serper API Key: For web search capabilities (used by the Research Crew). You can get a Serper API key at https://serper.dev/

Create a .env file in your project directory:

# Windows
echo NOTION_API_KEY=your-notion-api-key > .env
echo OPENAI_API_KEY=your-openai-api-key >> .env
echo SERPER_API_KEY=your-serper-api-key >> .env
echo DATABASE_ID=your-notion-database-id >> .env

# macOS/Linux
echo "NOTION_API_KEY=your-notion-api-key" > .env
echo "OPENAI_API_KEY=your-openai-api-key" >> .env
echo "SERPER_API_KEY=your-serper-api-key" >> .env
echo "DATABASE_ID=your-notion-database-id" >> .env

To get your DATABASE_ID, you'll need to:

1. Open your Notion database in the browser
2. Look at the URL, which will be something like: https://www.notion.so/workspace/83c75a39f33b4934b5a0d251f5882b52?v=...
3. The 32-character string after your workspace name and before the ? is your DATABASE_ID
4. Copy this ID and add it to your .env file

You'll also need to share your database with your integration:

1. Open your database in Notion
2. Click the "..." menu in the top right
3. Click "Add connections"
4. Find and select your integration from the list

Here's a useful tutorial from Notion on setting up your first integration.

The components

1. The Notion API client

A single class handles every call to Notion:

# orchestrator/notion_api.py
import os
import asyncio
from typing import Dict, List, Any, Optional
from notion_client import AsyncClient

class NotionAPI:
    def __init__(self, api_key: str = None, database_id: str = None):
        """Initialize the Notion API client."""
        self.api_key = api_key or os.environ.get("NOTION_API_KEY")
        self.database_id = database_id or os.environ.get("DATABASE_ID")
        self.client = AsyncClient(auth=self.api_key)

    async def query_database(self, filter_condition: Dict) -> List[Dict]:
        """Query the Notion database with a filter condition."""
        response = await self.client.databases.query(
            database_id=self.database_id,
            filter=filter_condition
        )
        return response.get("results", [])

    async def get_execute_tasks(self) -> List[Dict]:
        """Get all tasks with 'Execute' status."""
        filter_condition = {
            "property": "Status",
            "select": {
                "equals": "Execute"
            }
        }
        return await self.query_database(filter_condition)

    async def update_task_status(self, page_id: str, status: str) -> None:
        """Update the status of a task."""
        await self.client.pages.update(
            page_id=page_id,
            properties={
                "Status": {
                    "select": {
                        "name": status
                    }
                }
            }
        )

    async def update_task_response(self, page_id: str, response: str) -> None:
        """Update the response of a task."""
        # Split long text into chunks of 1900 characters (safety margin below 2000 limit)
        chunks = self._split_text_into_chunks(response, 1900)

        rich_text_array = []
        for chunk in chunks:
            rich_text_array.append({
                "type": "text",
                "text": {
                    "content": chunk
                }
            })

        await self.client.pages.update(
            page_id=page_id,
            properties={
                "Response": {
                    "rich_text": rich_text_array
                }
            }
        )

    async def update_page_content(self, page_id: str, content: str) -> None:
        """Update the content of a page with formatted blocks."""
        # First, retrieve existing children to avoid duplicating content
        existing = await self.client.blocks.children.list(block_id=page_id)

        # Delete existing children if any
        for block in existing.get("results", []):
            await self.client.blocks.delete(block_id=block["id"])

        # Create new content blocks
        blocks = []

        # Add a heading
        blocks.append({
            "object": "block",
            "type": "heading_2",
            "heading_2": {
                "rich_text": [{"type": "text", "text": {"content": "AI Response"}}]
            }
        })

        # Split content into paragraphs
        paragraphs = content.split("\n\n")
        for paragraph in paragraphs:
            if paragraph.strip():
                # For code blocks
                if paragraph.startswith("```") and paragraph.endswith("```"):
                    blocks.append({
                        "object": "block",
                        "type": "code",
                        "code": {
                            "rich_text": [{"type": "text", "text": {"content": paragraph[3:-3]}}],
                            "language": "plain text"
                        }
                    })
                else:
                    # Split long paragraphs into chunks of 1900 characters
                    chunks = self._split_text_into_chunks(paragraph, 1900)
                    for chunk in chunks:
                        blocks.append({
                            "object": "block",
                            "type": "paragraph",
                            "paragraph": {
                                "rich_text": [{"type": "text", "text": {"content": chunk}}]
                            }
                        })

        # Update the page with new blocks
        await self.client.blocks.children.append(
            block_id=page_id,
            children=blocks
        )

    def _split_text_into_chunks(self, text: str, chunk_size: int) -> List[str]:
        """Split text into chunks of specified size."""
        if not text:
            return [""]

        chunks = []
        for i in range(0, len(text), chunk_size):
            chunks.append(text[i:i + chunk_size])

        return chunks

That covers everything the system needs from Notion:

• Querying for tasks with "Execute" status
• Updating task status as we process it
• Writing the AI's response back to Notion

2. Building the task orchestrator

The orchestrator sits between Notion and the crews, and does the actual coordinating:

# orchestrator/orchestrator.py
import os
import logging
from typing import Dict, Any, List
from orchestrator.notion_api import NotionAPI
from orchestrator.crew_manager import CrewManager

logger = logging.getLogger(__name__)

class TaskOrchestrator:
    def __init__(self):
        """Initialize the task orchestrator."""
        self.notion_api = NotionAPI()
        self.crew_manager = CrewManager()

    async def process_execute_tasks(self):
        """Process all tasks with 'Execute' status."""
        # Get tasks with 'Execute' status
        tasks = await self.notion_api.get_execute_tasks()

        if not tasks:
            logger.info("No tasks to execute.")
            return

        logger.info(f"Found {len(tasks)} tasks to execute.")

        # Process each task
        for task in tasks:
            await self.process_task(task)

    async def process_task(self, task: Dict[str, Any]):
        """Process a single task."""
        page_id = task["id"]

        try:
            # Update status to In Progress
            await self.notion_api.update_task_status(page_id, "In Progress")

            # Extract task content
            task_content = self._extract_task_content(task)

            # Determine which crew should handle this task
            crew_name, crew_reasoning = await self.crew_manager.determine_crew(task_content)

            # Process the task with the appropriate crew
            result_data = await self.crew_manager.process_with_crew(crew_name, task_content)

            # For research crew, we get back both the result and thought process
            if crew_name == "research_crew" and isinstance(result_data, tuple):
                result, thought_logs = result_data
            else:
                result = result_data
                thought_logs = []

            # Update the task with the result
            await self.notion_api.update_task_response(page_id, result)

            # Update the page content with more detailed information
            content = f"Task processed by: {crew_name}\n\nReasoning: {crew_reasoning}\n\nResult:\n{result}"

            # Add thought process if available
            if thought_logs:
                content += "\n\n## Agent Thought Process\n\n"
                content += "```\n"
                content += "".join(thought_logs)
                content += "\n```"

            await self.notion_api.update_page_content(page_id, content)

            # Update status to Done
            await self.notion_api.update_task_status(page_id, "Done")

            logger.info(f"Successfully processed task {page_id}")

        except Exception as e:
            logger.error(f"Error processing task {page_id}: {str(e)}")
            # Update status to indicate error
            await self.notion_api.update_task_response(page_id, f"Error: {str(e)}")
            await self.notion_api.update_task_status(page_id, "Error")

    def _extract_task_content(self, task: Dict[str, Any]) -> str:
        """Extract the task content from the Notion page."""
        properties = task.get("properties", {})
        task_property = properties.get("Task", {})
        title = task_property.get("title", [])

        if title and len(title) > 0:
            return title[0].get("plain_text", "")

        return ""

Its workflow is:

1. Find tasks marked for execution
2. Update their status to "In Progress"
3. Determine which AI crew should handle each task
4. Process the task with the appropriate crew
5. Update the task with the results
6. Mark the task as "Done"

3. Creating the crew manager

The crew manager decides which crew handles a given task:

# orchestrator/crew_manager.py
import os
import logging
from typing import Tuple, Dict, Any
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage
from crews.research_crew.crew import ResearchCrew

logger = logging.getLogger(__name__)

class CrewManager:
    def __init__(self):
        """Initialize the crew manager."""
        self.llm = ChatOpenAI(
            api_key=os.environ.get("OPENAI_API_KEY"),
            model="gpt-4o-mini",
            temperature=0
        )

    async def determine_crew(self, task_content: str) -> Tuple[str, str]:
        """Determine which crew should handle this task."""
        system_prompt = """
        You are a task router that determines which specialized crew should handle a given task.
        Available crews:
        - research_crew: For tasks requiring web research, information gathering, and synthesis
        - default: For general tasks that don't fit other specialized crews

        Respond with ONLY the crew name followed by a brief explanation, like this:
        research_crew: This task requires gathering information from multiple sources
        OR
        default: This is a general task that doesn't require specialized handling
        """

        messages = [
            SystemMessage(content=system_prompt),
            HumanMessage(content=f"Task: {task_content}")
        ]

        response = self.llm.invoke(messages).content.strip()

        # Parse the response
        try:
            crew_name, reasoning = response.split(":", 1)
            crew_name = crew_name.strip().lower()
            reasoning = reasoning.strip()

            # Validate crew name
            if crew_name not in ["research_crew", "default"]:
                logger.warning(f"Invalid crew name: {crew_name}. Falling back to default.")
                crew_name = "default"
                reasoning = "Fallback due to invalid crew determination."

            return crew_name, reasoning

        except Exception as e:
            logger.error(f"Error parsing crew determination: {str(e)}")
            return "default", "Fallback due to error in crew determination."

    async def process_with_crew(self, crew_name: str, task_content: str):
        """Process the task with the appropriate crew."""
        if crew_name == "research_crew":
            try:
                crew = ResearchCrew()
                result, thought_logs = await crew.run(task_content)
                return result, thought_logs
            except Exception as e:
                logger.error(f"Error with research crew: {str(e)}")
                # Fall back to default processing
                return await self._process_with_default(task_content)
        else:
            return await self._process_with_default(task_content)

    async def _process_with_default(self, task_content: str) -> str:
        """Process the task with the default OpenAI processing."""
        messages = [
            SystemMessage(content="You are a helpful assistant that processes tasks."),
            HumanMessage(content=f"Task: {task_content}")
        ]

        response = self.llm.invoke(messages).content.strip()
        return response

It does two things:

1. It determines which crew should handle a particular task based on its content
2. It delegates the task to the appropriate crew for processing

The reason for splitting it out is that a new specialized crew can be added without touching the orchestration logic.

4. Implementing the research crew

Here is a Research Crew built with CrewAI. It is deliberately simplified; the CrewAI documentation covers the more elaborate cases.

# crews/research_crew/crew.py
import os
from typing import Dict, Any, List
from crewai import Crew, Agent, Task, Process
from pydantic import BaseModel, Field
from crewai_tools import SerperDevTool, WebsiteSearchTool, ScrapeWebsiteTool

class InitialResearchOutput(BaseModel):
    """Output model for initial research task"""
    key_facts: List[str] = Field(..., description="Key facts and data points discovered")
    recent_developments: List[str] = Field(..., description="Recent developments in the topic")
    expert_opinions: List[Dict[str, str]] = Field(..., description="Expert opinions with source attribution")
    statistics: List[Dict[str, str]] = Field(..., description="Relevant statistics with source attribution")
    sources: List[str] = Field(..., description="List of sources consulted")

class ResearchOutput(BaseModel):
    """Output model for final research analysis"""
    executive_summary: str = Field(..., description="Brief overview of the entire analysis")
    initial_research: InitialResearchOutput = Field(..., description="Raw research findings")
    key_findings: List[str] = Field(..., description="Key findings from the analysis")
    trend_analysis: Dict[str, str] = Field(..., description="Analysis of identified trends")
    recommendations: List[str] = Field(..., description="Actionable recommendations based on the research")

# Global variable to store thought process
thought_process_logs = []

def callback_function(output):
    """Callback function to track agent's thought process"""
    global thought_process_logs

    if hasattr(output, 'output'):
        output_text = f"\nTask completed!\nOutput: {output.output}\n"
    else:
        output_text = f"\nTool used: {str(output)}\n"

    # Truncate extremely long thought entries to prevent issues
    if len(output_text) > 10000:
        truncated_text = output_text[:10000] + "... [truncated due to length]"
        print(f"Truncated extremely long thought process entry ({len(output_text)} chars)")
        output_text = truncated_text

    # Add to global logs
    thought_process_logs.append(output_text)
    print(output_text)

class ResearchCrew:
    def __init__(self):
        """Initialize the research crew."""
        # Initialize tools
        try:
            self.search_tool = SerperDevTool(api_key=os.environ.get("SERPER_API_KEY"))
            self.website_search_tool = WebsiteSearchTool()
            self.scrape_tool = ScrapeWebsiteTool()

            tools = [self.search_tool, self.website_search_tool, self.scrape_tool]
        except Exception as e:
            print(f"Error initializing tools: {str(e)}")
            # Fallback to simpler tools if needed
            tools = []

        # Create agents
        self.researcher = Agent(
            role="Research Analyst",
            goal="Find accurate and relevant information on the web",
            backstory="You are an expert at finding information online and extracting key insights.",
            verbose=True,
            tools=tools,
            allow_delegation=False,
            step_callback=callback_function,
            output_pydantic=InitialResearchOutput
        )

        self.senior_researcher = Agent(
            role="Senior Research Analyst",
            goal="Synthesize research findings into comprehensive reports",
            backstory="You are an expert at analyzing information and creating insightful summaries.",
            verbose=True,
            allow_delegation=False,
            step_callback=callback_function,
            output_pydantic=ResearchOutput
        )

    async def run(self, task_content: str) -> str:
        """Run the research crew on the given task."""
        global thought_process_logs
        thought_process_logs = []  # Reset logs for new run

        # Create tasks
        research_task = Task(
            description=f"Research the following topic thoroughly: {task_content}",
            expected_output="Detailed research findings with sources",
            agent=self.researcher
        )

        analysis_task = Task(
            description="Analyze the research findings and create a comprehensive report",
            expected_output="A well-structured report with key insights and conclusions",
            agent=self.senior_researcher,
            context=[research_task]
        )

        # Create crew
        crew = Crew(
            agents=[self.researcher, self.senior_researcher],
            tasks=[research_task, analysis_task],
            verbose=True,
            process=Process.sequential
        )

        # Run the crew
        result = crew.kickoff()

        # Return both the result and the thought process logs
        return result, thought_process_logs

The Research Crew consists of two agents:

1. A Research Analyst who finds information online using search tools
2. A Senior Research Analyst who turns those findings into a written report

Splitting it this way keeps gathering separate from synthesis, which are different jobs and tend to go better with different prompts.

Recording how the agents got there

A callback on the agents captures their intermediate steps while a task runs, and writes them onto the Notion page alongside the result.

The callback looks like this:

# Global variable to store thought process
thought_process_logs = []

def callback_function(output):
    """Callback function to track agent's thought process"""
    global thought_process_logs

    if hasattr(output, 'output'):
        output_text = f"\nTask completed!\nOutput: {output.output}\n"
    else:
        output_text = f"\nTool used: {str(output)}\n"

    # Add to global logs
    thought_process_logs.append(output_text)
    print(output_text)

It fires at every step and records:

• When tools are being used and what they're being used for
• The intermediate thinking steps of each agent
• When tasks are completed and what their outputs are

Those logs land on the Notion page next to the answer:

I keep them for four reasons:

1. Debugging: You can see exactly where things might have gone wrong
2. Learning: You gain insight into how AI approaches different types of tasks
3. Trust: Users can verify the sources and reasoning behind conclusions
4. Improvement: You can identify patterns that lead to better or worse results

This is what I mean by transparent orchestration in OPERA: the assistance is an observable process sitting in the workspace rather than a black box you accept output from.

Notion's character limit

Rich text blocks in the Notion API cap at 2000 characters, and generated content runs past that regularly, so it has to be chunked:

def _split_text_into_chunks(self, text: str, chunk_size: int) -> List[str]:
    """Split text into chunks of specified size."""
    if not text:
        return [""]

    chunks = []
    for i in range(0, len(text), chunk_size):
        chunks.append(text[i:i + chunk_size])

    return chunks

The helper splits text at 1900 characters, leaving a margin under the limit, so a long research report or a verbose thought log still saves in full.

The callback needs the same treatment for very long entries, which would otherwise cost memory or hit the API limit:

# Truncate extremely long thought entries to prevent issues
if len(output_text) > 10000:
    truncated_text = output_text[:10000] + "... [truncated due to length]"
    print(f"Truncated extremely long thought process entry ({len(output_text)} chars)")
    output_text = truncated_text

Neither is interesting, and both will bite you on the first real task.

5. Creating the scheduled service

The last piece is the service that keeps checking for work:

# scheduled_service.py
import os
import asyncio
import logging
from dotenv import load_dotenv
from orchestrator.orchestrator import TaskOrchestrator

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()

async def process_tasks():
    """Process tasks from Notion database."""
    try:
        orchestrator = TaskOrchestrator()
        await orchestrator.process_execute_tasks()
    except Exception as e:
        logger.error(f"Error processing tasks: {str(e)}")
        raise

async def scheduled_task():
    """Run the task processing on a schedule."""
    while True:
        logger.info("Checking for tasks to process...")
        await process_tasks()
        logger.info("Task processing complete. Waiting for next cycle...")
        # Wait for 5 minutes before checking again
        await asyncio.sleep(300)  # 300 seconds = 5 minutes

if __name__ == "__main__":
    logger.info("Starting scheduled service...")
    asyncio.run(scheduled_task())

It loops every five minutes as a background process, so tasks get picked up whether or not you are watching.

Deploying to Render

Three files prepare the project for Render:

1. Create a requirements.txt file:

fastapi==0.104.1
uvicorn==0.23.2
notion-client==2.0.0
openai==1.3.5
crewai==0.28.0
crewai-tools==0.1.11
python-dotenv==1.0.0
langchain==0.0.335
langchain-openai==0.0.2

1. Create a Procfile for Render:

web: python scheduled_service.py

1. Create a runtime.txt file:

python-3.11.6

Then deploy:

1. Sign up for a Render account at https://render.com/
2. Connect your GitHub repository
3. Create a new Web Service
4. Select your repository
5. Configure it: name notion-agent-integration, environment Python, build command pip install -r requirements.txt, start command python scheduled_service.py
6. Add the environment variables NOTION_API_KEY, OPENAI_API_KEY, SERPER_API_KEY and DATABASE_ID
7. Pick a plan (the free tier is enough for personal use) and click "Create Web Service"

Learn more about how to setup Render here.

Seeing it in action

Create a task in Notion to test it:

1. Create a new entry in your Notion database
2. Set the Status to "Execute"
3. Add a task description, such as "Research the latest developments in AI and summarize the key trends"
4. Wait for the next polling cycle (up to 5 minutes)

You should see the task status change to "In Progress" and then to "Done" once processed, with the Response field populated with the AI's output.

What this shows about OPERA

Four things in this build are the Operations layer doing what I claimed it would:

1. It fits the existing structure: the system works inside your Notion setup and respects the boundaries between Projects, Environments, Resources and Archives.
2. It routes rather than executes: the layer sends each task to a specialized crew, which is the coordination OPERA is built around.
3. It needs no upkeep once running: this was one of the four weaknesses I found in my own PARA setup.
4. It reads and writes the knowledge base: results go back where the source material already lives, so the base improves as it is used.

What comes next

Part 2 extends the system in four directions:

1. A Content Creation Crew for blog posts and social content
2. An interactive review process that allows users to add comments to Notion pages and get new iterations of the results
3. More context, by holding additional information in the Notion database so results have more to work from
4. Dynamic agent creation, through an Agent database in Notion, so crews no longer have to be defined by hand in the backend

Each one moves more of the configuration out of the code and into the workspace, where I can change it without a deploy.

Where that leaves the framework

A framework earns its keep at the point where it turns into something that runs, and this integration is the first evidence I have that OPERA describes a workable system rather than a tidy diagram.

The part I like most is how little it takes. There is no specialist knowledge and no enterprise budget behind any of this.

That is the point of Framework Reboot: serious productivity systems for individuals and small organizations, who otherwise sit between a basic productivity app and an enterprise AI platform with nothing in between.

The code here is simplified for readability. If you hit problems setting it up, Cursor or a similar AI-assisted editor will get you through most of them faster than I can.