Building a MCP-Powered Task Manager Agent with Agno and Supabase: A Step-by-Step Guide

import os
import asyncio
from textwrap import dedent
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.mcp import MCPTools

# Check if required environment variables are available
if "SUPABASE_ACCESS_TOKEN" not in os.environ:
    raise ValueError("SUPABASE_ACCESS_TOKEN environment variable is not set. Please set it and try again.")

if "OPENAI_API_KEY" not in os.environ:
    raise ValueError("OPENAI_API_KEY environment variable is not set. Please set it and try again.")

Step 3: Understanding the Model Context Protocol (MCP)

async def main():
    # Using async context manager as specified in the docs
    command = f"npx -y @supabase/mcp-server-supabase@latest --access-token {os.environ['SUPABASE_ACCESS_TOKEN']}"
    async with MCPTools(command) as mcp_tools:
        # Create agent with these tools
        # ...

Step 5: Creating a Clean User Interface

The user interface is simple but effective:

def print_welcome():
    print(f'''
Agno Task Manager Agent (Supabase MCP)
--------------------------------------
You are now chatting with an AI agent that can manage your tasks in Supabase.

You can:
- Add new tasks (e.g. "Add a task for John to finish the report by Friday, 80% done.")
- Update tasks (e.g. "Change the due date for the report task to next Monday.")
- Remove tasks (e.g. "Delete the task about the report.")
- List all tasks (e.g. "Show me all my tasks.")

Just type your requests in natural language. Type 'exit' or 'quit' to stop.
''')

# In the main loop:
while True:
    user_input = input("\nYou: ").strip()
    if user_input.lower() in {"exit", "quit"}:
        print("Exiting. Your tasks are saved in Supabase!")
        break
    
    try:
        await agent.aprint_response(user_input, stream=True)
    except Exception as e:
        print(f"\n❌ Error during agent response: {str(e)}")

Notice that we use await agent.aprint_response() instead of the synchronous version. This is because our entire application uses the asyncio pattern, which is essential when working with MCP tools.

Step 6: The Complete Async Implementation

Here's the complete implementation, with special attention to async patterns:

import os
import asyncio
from textwrap import dedent
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.mcp import MCPTools

# Check if required environment variables are available
if "SUPABASE_ACCESS_TOKEN" not in os.environ:
    raise ValueError("SUPABASE_ACCESS_TOKEN environment variable is not set. Please set it and try again.")

if "OPENAI_API_KEY" not in os.environ:
    raise ValueError("OPENAI_API_KEY environment variable is not set. Please set it and try again.")

def print_welcome():
    print(f'''
Agno Task Manager Agent (Supabase MCP)
--------------------------------------
You are now chatting with an AI agent that can manage your tasks in Supabase.

You can:
- Add new tasks (e.g. "Add a task for John to finish the report by Friday, 80% done.")
- Update tasks (e.g. "Change the due date for the report task to next Monday.")
- Remove tasks (e.g. "Delete the task about the report.")
- List all tasks (e.g. "Show me all my tasks.")

Just type your requests in natural language. Type 'exit' or 'quit' to stop.
''')

async def main():
    """Main async function to run the agent."""
    print_welcome()
    
    try:
        # Initialize MCP tools using the context manager pattern
        command = f"npx -y @supabase/mcp-server-supabase@latest --access-token {os.environ['SUPABASE_ACCESS_TOKEN']}"
        async with MCPTools(command) as mcp_tools:
            # Create the agent with initialized tools
            agent = Agent(
                name="Agno Supabase Task Manager",
                role="Task manager agent that stores tasks in Supabase via MCP.",
                model=OpenAIChat(id="gpt-4o-mini"),
                tools=[mcp_tools],
                instructions=dedent("""
                    You are a task manager agent that uses Supabase MCP to store tasks. Each task has:
                    - description (text)
                    - completion (percent)
                    - responsible (person)
                    - due_date (YYYY-MM-DD)
                    
                    Follow these exact steps to interact with Supabase:
                    
                    1. First, use `list_projects` to find all available Supabase projects
                    
                    2. Look for the project named "AgnoTaskList" and get its project ID (ref)
                    
                    3. For all database operations, specify the project ID explicitly:
                       - To check tables: use `list_tables` with the project ID and schema="public"
                       - To create table: use `apply_migration` with the project ID
                       - To manipulate data: use `execute_sql` with the project ID
                    
                    4. If the TaskList table doesn't exist, create it with `apply_migration`:
                       ```sql
                       CREATE TABLE IF NOT EXISTS TaskList (
                           id SERIAL PRIMARY KEY,
                           description TEXT NOT NULL,
                           completion INTEGER NOT NULL,
                           responsible TEXT NOT NULL,
                           due_date DATE NOT NULL
                       );
                       ```
                       
                    5. For regular operations, use `execute_sql`:
                       - To add tasks: Use the current date (CURRENT_DATE) as the default due date if none is specified
                       - To list tasks: `SELECT * FROM TaskList;`
                       - To update tasks: `UPDATE TaskList SET column_name = new_value WHERE condition RETURNING *;`
                       - To delete tasks: `DELETE FROM TaskList WHERE condition RETURNING *;`
                    
                    IMPORTANT RULES:
                    - Always include the project ID in ALL database operations
                    - If no due date is specified when adding a task, use the current date (CURRENT_DATE)
                    - Show the actual results returned from database operations
                    - When listing tasks, present them in a clear, formatted table
                    - If any tool call fails, show the error message and try to troubleshoot it
                """),
                show_tool_calls=True,
                add_state_in_messages=True,
                markdown=True,
            )
            
            # Interact with the agent
            while True:
                user_input = input("\nYou: ").strip()
                if user_input.lower() in {"exit", "quit"}:
                    print("Exiting. Your tasks are saved in Supabase!")
                    break
                
                try:
                    await agent.aprint_response(user_input, stream=True)
                except Exception as e:
                    print(f"\n❌ Error during agent response: {str(e)}")
    
    except Exception as e:
        print(f"\n❌ ERROR: {str(e)}")

if __name__ == "__main__":
    try:
        # Run the async main function
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nOperation interrupted by user. Exiting...")
    except Exception as e:
        print(f"\n❌ ERROR: {str(e)}")

1. Entry Point: asyncio.run(main()) is our entry point, which sets up the event loop.
2. Async Context Manager: async with MCPTools(...) as mcp_tools: ensures proper initialization and cleanup of resources.
3. Agent Response: await agent.aprint_response(...) pauses execution while waiting for the agent to process and respond.