adding

agents/__init__.py

@@ -0,0 +1,5 @@
+"""OpenAI Agents SDK orchestration for Todo AI Chatbot."""
+
+from src.agents.todo_agent import TodoAgent, create_todo_agent
+
+__all__ = ["TodoAgent", "create_todo_agent"]

agents/todo_agent.py

@@ -0,0 +1,347 @@
+"""
+TodoAgent - OpenAI Agent orchestration with MCP tools.
+
+Per @specs/001-chatbot-mcp/plan.md Section VIII - AI Chatbot Architecture
+OpenAI Agents SDK for orchestration with MCP tool calling.
+"""
+from openai import OpenAI
+from typing import List, Dict, Any, Optional, AsyncGenerator
+from uuid import UUID
+import logging
+import json
+
+from src.config import settings
+from src.mcp.server import get_mcp_server
+
+logger = logging.getLogger(__name__)
+
+
+class TodoAgent:
+    """
+    OpenAI Agent for Todo task management with MCP tool integration.
+
+    This agent:
+    - Uses OpenAI Chat Completions API with tool calling
+    - Integrates with MCP tools for task operations
+    - Maintains conversation context for multi-turn dialogs
+    - Enforces user isolation by passing user_id to all tool calls
+    """
+
+    # System prompt for friendly, concise, task-oriented behavior
+    SYSTEM_PROMPT = """You are a helpful task management assistant for TaskFlow. Your role is to help users manage their todo tasks through natural language.
+
+Key behaviors:
+- Be friendly and concise in your responses
+- Focus on helping users create, view, complete, and delete tasks
+- When users ask to add tasks, extract the task title and any description
+- When users ask to see tasks, list their tasks clearly with numbers for easy reference
+- When users ask to complete/delete/update tasks, ALWAYS call list_tasks FIRST to identify the task
+- Users may refer to tasks by number (e.g., "first task", "task 2") or by title/content
+- Use the available tools to perform all task operations
+- Never make up task information - always use the tools
+- The user_id parameter is automatically injected - do NOT ask the user for it
+
+Task identification workflow:
+1. When user wants to update/delete/complete a task, FIRST call list_tasks
+2. Show the user the tasks with numbers: "1. Task title", "2. Task title", etc.
+3. If user specified a number or title, match it to get the task_id
+4. Then call the appropriate tool (update_task, delete_task, complete_task) with the task_id
+
+Available tools:
+- add_task: Create a new task with title and optional description
+- list_tasks: Show all tasks with their IDs (call this FIRST before update/delete/complete)
+- complete_task: Mark a task as completed (requires task_id from list_tasks)
+- delete_task: Remove a task permanently (requires task_id from list_tasks)
+- update_task: Change a task's title or description (requires task_id from list_tasks)
+
+Note: user_id is automatically provided for all tool calls. Never ask the user for their user ID."""
+
+    def __init__(self, user_id: UUID):
+        """
+        Initialize the TodoAgent for a specific user.
+
+        Args:
+            user_id: The user's UUID for data isolation
+        """
+        self.user_id = user_id
+
+        # Log API key info for debugging (don't log full key)
+        key_preview = settings.openai_api_key[:10] if settings.openai_api_key else "None"
+        key_length = len(settings.openai_api_key) if settings.openai_api_key else 0
+        logger.info(f"TodoAgent initializing with OpenAI API key: {key_preview}... (length: {key_length})")
+
+        self.client = OpenAI(api_key=settings.openai_api_key)
+        self.model = settings.openai_model
+
+        # Get MCP server and register tools as OpenAI functions
+        self.mcp_server = get_mcp_server()
+        self.tools = self._get_openai_tools()
+
+        logger.info(f"TodoAgent initialized for user {user_id}")
+
+    def _get_openai_tools(self) -> List[Dict[str, Any]]:
+        """
+        Convert MCP tools to OpenAI function format.
+
+        Returns:
+            List of tool definitions in OpenAI format
+
+        Per @specs/001-chatbot-mcp/contracts/mcp-tools.json
+        """
+        tools = []
+
+        # Get all registered MCP tools from SimpleMCPRegistry
+        mcp_tools = self.mcp_server.list_tools()
+
+        for tool in mcp_tools:
+            # Convert MCP tool schema to OpenAI function format
+            function_def = {
+                "type": "function",
+                "function": {
+                    "name": tool.name,
+                    "description": tool.description,
+                    "parameters": tool.parameters.copy()  # Copy to avoid mutation
+                }
+            }
+
+            # Add user_id parameter to all tools for data isolation
+            # Note: user_id is auto-filled and should NOT be in required parameters
+            if "properties" not in function_def["function"]["parameters"]:
+                function_def["function"]["parameters"]["properties"] = {}
+
+            function_def["function"]["parameters"]["properties"]["user_id"] = {
+                "type": "string",
+                "description": "User ID for data isolation (auto-filled, do not ask user)",
+            }
+
+            # Initialize required array if not present
+            if "required" not in function_def["function"]["parameters"]:
+                function_def["function"]["parameters"]["required"] = []
+
+            # IMPORTANT: Do NOT add user_id to required - it's auto-injected
+            # This prevents the model from asking the user for their user_id
+
+            tools.append(function_def)
+
+        logger.info(f"Registered {len(tools)} tools with OpenAI agent")
+        return tools
+
+    async def process_message(
+        self,
+        user_message: str,
+        conversation_history: Optional[List[Dict[str, str]]] = None
+    ) -> AsyncGenerator[str, None]:
+        """
+        Process a user message through the agent with tool execution.
+
+        Args:
+            user_message: The user's message text
+            conversation_history: Previous messages in the conversation
+
+        Yields:
+            Response text chunks as they're generated
+
+        Per @specs/001-chatbot-mcp/plan.md - MCP First with OpenAI Agents
+        """
+        # Build messages array
+        messages = [
+            {"role": "system", "content": self.SYSTEM_PROMPT}
+        ]
+
+        # Add conversation history if provided
+        if conversation_history:
+            messages.extend(conversation_history)
+
+        # Add current user message
+        messages.append({"role": "user", "content": user_message})
+
+        logger.info(f"Processing message for user {self.user_id}: {user_message[:50]}...")
+
+        # Check if this is an update/delete/complete request
+        # If so, pre-load tasks to provide context
+        update_delete_keywords = ["update", "delete", "remove", "change", "modify", "complete", "finish", "mark"]
+        needs_task_list = any(keyword in user_message.lower() for keyword in update_delete_keywords)
+
+        if needs_task_list:
+            # Get tasks first and add to context
+            list_tool = self.mcp_server.get_tool("list_tasks")
+            if list_tool:
+                tasks_result = await list_tool.handler(user_id=str(self.user_id), include_completed=True)
+                if tasks_result.get("success") and tasks_result.get("tasks"):
+                    # Add task context to system prompt
+                    task_list = "\n".join([
+                        f"Task {i+1}: ID={t['id']}, Title='{t['title']}', Completed={t['completed']}"
+                        for i, t in enumerate(tasks_result["tasks"])
+                    ])
+                    enhanced_prompt = self.SYSTEM_PROMPT + f"\n\nCURRENT USER TASKS:\n{task_list}\n\nWhen user refers to a task by number or title, use the corresponding ID from this list."
+                    messages[0] = {"role": "system", "content": enhanced_prompt}
+                    logger.info(f"Pre-loaded {len(tasks_result['tasks'])} tasks for context")
+
+        try:
+            # Make chat completion request with tools
+            response = self.client.chat.completions.create(
+                model=self.model,
+                messages=messages,
+                tools=self.tools,
+                tool_choice="auto",  # Let model decide when to use tools
+                temperature=0.7,  # Slightly creative but focused
+                max_tokens=1000,  # Reasonable response length
+            )
+
+            # Process response
+            choice = response.choices[0]
+            message = choice.message
+
+            # Check if model wants to call tools
+            if message.tool_calls:
+                # Execute tool calls and collect results
+                tool_messages = []
+                for tool_call in message.tool_calls:
+                    result = await self._execute_tool_call(tool_call)
+                    # Add tool result as a tool message
+                    tool_messages.append({
+                        "role": "tool",
+                        "tool_call_id": tool_call.id,
+                        "content": result
+                    })
+
+                # Add assistant message with tool calls
+                messages.append({
+                    "role": "assistant",
+                    "content": message.content or "",
+                    "tool_calls": message.tool_calls
+                })
+
+                # Add tool result messages
+                messages.extend(tool_messages)
+
+                # Get follow-up response with tool results
+                follow_up = self.client.chat.completions.create(
+                    model=self.model,
+                    messages=messages,
+                    temperature=0.7,
+                    max_tokens=1000,
+                )
+
+                if follow_up.choices[0].message.content:
+                    yield follow_up.choices[0].message.content
+
+            # Direct text response (no tools needed)
+            elif message.content:
+                yield message.content
+
+            else:
+                yield "I understand. How can I help you with your tasks?"
+
+        except Exception as e:
+            error_type = type(e).__name__
+            error_msg = str(e)
+            logger.error(f"Error processing message: {error_type}: {error_msg}", exc_info=True)
+
+            # Provide more helpful error messages
+            if "Connection" in error_msg or "connect" in error_msg.lower():
+                yield "I'm having trouble connecting to my AI service. Please check if the OpenAI API key is configured correctly in Railway environment variables."
+            elif "401" in error_msg or "Unauthorized" in error_msg or "authentication" in error_msg.lower():
+                yield "My AI service credentials are invalid. Please check the OpenAI API key in Railway environment variables."
+            elif "rate" in error_msg.lower() or "limit" in error_msg.lower():
+                yield "I've reached my rate limit. Please try again in a moment."
+            else:
+                yield f"I encountered an error ({error_type}): {error_msg}"
+
+    async def _execute_tool_call(self, tool_call) -> str:
+        """
+        Execute a single tool call from OpenAI.
+
+        Args:
+            tool_call: The OpenAI tool call object
+
+        Returns:
+            Result message to display to user
+
+        Per @specs/001-chatbot-mcp/plan.md - MCP First architecture
+        """
+        function_name = tool_call.function.name
+        function_args = json.loads(tool_call.function.arguments)
+
+        # Inject user_id for data isolation
+        function_args["user_id"] = str(self.user_id)
+
+        logger.info(f"Executing tool: {function_name} with args: {function_args}")
+
+        try:
+            # Get the tool from SimpleMCPRegistry
+            tool = self.mcp_server.get_tool(function_name)
+
+            if not tool:
+                return f"Error: Tool '{function_name}' not found"
+
+            # For update_task and delete_task, validate task_id exists first
+            if function_name in ["update_task", "delete_task", "complete_task"]:
+                task_id = function_args.get("task_id")
+                if task_id:
+                    # Verify the task exists and belongs to user before proceeding
+                    list_tool = self.mcp_server.get_tool("list_tasks")
+                    tasks_result = await list_tool.handler(user_id=str(self.user_id), include_completed=True)
+                    valid_task_ids = [t["id"] for t in tasks_result.get("tasks", [])]
+
+                    if task_id not in valid_task_ids:
+                        # Task doesn't exist or doesn't belong to user
+                        # Provide helpful error with current tasks
+                        if tasks_result.get("tasks"):
+                            task_list = "\n".join([
+                                f"Task {i+1}: {t['title']}"
+                                for i, t in enumerate(tasks_result["tasks"])
+                            ])
+                            return f"Error: Task not found. Here are your current tasks:\n{task_list}\n\nPlease specify which task you'd like to {function_name.replace('_', ' ')}."
+                        else:
+                            return "Error: You don't have any tasks yet. Create some tasks first!"
+
+            # Execute the tool via MCP
+            result = await tool.handler(**function_args)
+
+            # Parse result
+            if isinstance(result, dict):
+                if result.get("success"):
+                    # Format success message based on tool
+                    if function_name == "add_task":
+                        return f"✓ Task '{result.get('title')}' created successfully!"
+                    elif function_name == "complete_task":
+                        return f"✓ Task '{result.get('title')}' marked as complete!"
+                    elif function_name == "delete_task":
+                        return f"✓ Task '{result.get('title')}' deleted."
+                    elif function_name == "update_task":
+                        return f"✓ Task updated successfully!"
+                    elif function_name == "list_tasks":
+                        tasks = result.get("tasks", [])
+                        count = result.get("count", 0)
+                        if count == 0:
+                            return "You don't have any tasks yet."
+                        # Number tasks for easy reference
+                        task_list = "\n".join([
+                            f"{i+1}. {t['title']}" + (" ✓" if t['completed'] else "")
+                            for i, t in enumerate(tasks)
+                        ])
+                        return f"You have {count} task(s):\n{task_list}\n\nYou can refer to tasks by number (e.g., \"complete task 1\")"
+                    else:
+                        return "Operation completed successfully!"
+                else:
+                    return f"Error: {result.get('error', 'Unknown error')}"
+
+            return str(result)
+
+        except Exception as e:
+            logger.error(f"Error executing tool {function_name}: {e}")
+            return f"Error executing {function_name}: {str(e)}"
+
+
+def create_todo_agent(user_id: UUID) -> TodoAgent:
+    """
+    Factory function to create a TodoAgent instance.
+
+    Args:
+        user_id: The user's UUID
+
+    Returns:
+        Initialized TodoAgent instance
+    """
+    return TodoAgent(user_id)

api/__init__.py

@@ -0,0 +1,4 @@
+"""API package."""
+from src.api.dependencies import get_current_user, verify_user_ownership
+
+__all__ = ["get_current_user", "verify_user_ownership"]

api/dependencies.py

@@ -0,0 +1,69 @@
+"""
+FastAPI dependencies for authentication and authorization.
+
+Per @specs/001-auth-api-bridge/research.md - FastAPI Dependencies pattern
+"""
+from fastapi import Depends, HTTPException, Request, status
+from src.services.auth import verify_token
+
+
+async def get_current_user(request: Request) -> str:
+    """
+    Extract and verify user_id from JWT token.
+
+    This dependency enforces JWT authentication on protected endpoints.
+    Per @specs/001-auth-api-bridge/research.md
+
+    Args:
+        request: FastAPI request object
+
+    Returns:
+        User ID (UUID string) from verified JWT token
+
+    Raises:
+        HTTPException 401: If token is missing, invalid, or expired
+    """
+    auth_header = request.headers.get("Authorization")
+
+    if not auth_header or not auth_header.startswith("Bearer "):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Missing or invalid Authorization header",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    token = auth_header.split(" ")[1]
+    user_id = await verify_token(token)
+
+    if user_id is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid or expired token",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    # Attach user_id to request state for use in route handlers
+    request.state.user_id = user_id
+    return user_id
+
+
+async def verify_user_ownership(request: Request, user_id: str) -> None:
+    """
+    Verify that the requested user_id matches the authenticated user.
+
+    This enforces user isolation - users can only access their own resources.
+    Per @specs/001-auth-api-bridge/api/rest-endpoints.md security requirements
+
+    Args:
+        request: FastAPI request object (contains authenticated user_id in state)
+        user_id: User ID from URL path parameter
+
+    Raises:
+        HTTPException 403: If user_id doesn't match authenticated user
+    """
+    current_user = request.state.user_id
+    if current_user != user_id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Access forbidden: You can only access your own resources"
+        )

api/routes/__init__.py

@@ -0,0 +1,6 @@
+"""Routes package."""
+from src.api.routes.tasks import router as tasks_router
+from src.api.routes.health import router as health_router
+from src.api.routes.chat import router as chat_router
+
+__all__ = ["tasks_router", "health_router", "chat_router"]

api/routes/chat.py

@@ -0,0 +1,268 @@
+"""
+Chat API endpoints for Todo AI Chatbot.
+
+Per @specs/001-chatbot-mcp/contracts/openapi.yaml and @specs/001-chatbot-mcp/plan.md
+"""
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from sqlmodel import Session
+from uuid import UUID
+from datetime import datetime
+import time
+import logging
+
+from src.config import engine
+from src.api.dependencies import get_current_user
+from src.api.schemas.chat import ChatRequest, ChatResponse, Message, ChatError
+from src.services.chat import ChatService
+from src.agents.todo_agent import create_todo_agent
+from src.models.message import MessageRole
+
+logger = logging.getLogger(__name__)
+
+# Create router
+router = APIRouter(prefix="/api", tags=["chat"])
+
+
+@router.post("/{user_id}/chat", response_model=ChatResponse)
+async def chat_endpoint(
+    user_id: str,
+    request: ChatRequest,
+    http_request: Request,
+    current_user: str = Depends(get_current_user)
+):
+    """
+    Chat endpoint for AI-powered task management.
+
+    Processes user messages through OpenAI agent with MCP tool integration.
+    Creates new conversations or continues existing ones.
+
+    Per @specs/001-chatbot-mcp/plan.md:
+    - Stateless architecture: history loaded from DB each request
+    - MCP First: all task operations through MCP tools
+    - Data isolation: all queries filter by user_id
+
+    Per @specs/001-chatbot-mcp/contracts/openapi.yaml
+    """
+    start_time = time.time()
+
+    # Verify user ownership
+    if current_user != user_id:
+        logger.warning(f"User {current_user} attempted to access user {user_id} chat")
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Access forbidden: You can only access your own chat"
+        )
+
+    # Parse user_id as UUID
+    try:
+        user_uuid = UUID(user_id)
+    except ValueError:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid user ID format"
+        )
+
+    with Session(engine) as session:
+        try:
+            # Get or create conversation
+            conversation = None
+            conversation_id = request.conversation_id
+
+            if conversation_id:
+                # Validate user owns this conversation
+                conversation = ChatService.get_conversation(session, conversation_id, user_uuid)
+                if not conversation:
+                    raise HTTPException(
+                        status_code=status.HTTP_404_NOT_FOUND,
+                        detail="Conversation not found or access denied"
+                    )
+            else:
+                # Create new conversation for first message
+                conversation = ChatService.create_conversation(
+                    session=session,
+                    user_id=user_uuid,
+                    title="New Chat"  # Can be updated based on first message content
+                )
+                conversation_id = conversation.id
+                logger.info(f"Created new conversation {conversation_id} for user {user_id}")
+
+            # Sanitize user input
+            sanitized_message = ChatService.sanitize_user_input(request.message)
+
+            # Store user message
+            user_message = ChatService.store_message(
+                session=session,
+                conversation_id=conversation_id,
+                role=MessageRole.USER,
+                content=sanitized_message
+            )
+
+            # Load conversation history
+            history = ChatService.get_conversation_history(
+                session=session,
+                conversation_id=conversation_id,
+                user_id=user_uuid
+            )
+
+            # Format for OpenAI (exclude the message we just added)
+            formatted_history = ChatService.format_messages_for_openai(history[:-1])
+
+            # Create agent and process message
+            agent = create_todo_agent(user_uuid)
+
+            # Collect agent response
+            response_parts = []
+            async for chunk in agent.process_message(sanitized_message, formatted_history):
+                response_parts.append(chunk)
+
+            assistant_response = "".join(response_parts)
+
+            # Store assistant response
+            assistant_message = ChatService.store_message(
+                session=session,
+                conversation_id=conversation_id,
+                role=MessageRole.ASSISTANT,
+                content=assistant_response,
+                metadata={"processing_time": time.time() - start_time}
+            )
+
+            # Calculate processing time
+            processing_time = time.time() - start_time
+
+            # Log request
+            logger.info(
+                f"Chat processed: user={user_id}, "
+                f"conversation={conversation_id}, "
+                f"processing_time={processing_time:.2f}s, "
+                f"message_length={len(request.message)}"
+            )
+
+            # Build response
+            return ChatResponse(
+                conversation_id=conversation_id,
+                message=Message(
+                    id=assistant_message.id,
+                    role="assistant",
+                    content=assistant_response,
+                    created_at=assistant_message.created_at
+                ),
+                tasks=None  # Could be populated with affected tasks if needed
+            )
+
+        except HTTPException:
+            # Re-raise HTTP exceptions as-is
+            raise
+
+        except Exception as e:
+            # Log error for debugging
+            logger.error(f"Error processing chat request: {e}", exc_info=True)
+
+            # Return user-friendly error message
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail={
+                    "error": "Failed to process chat message",
+                    "message": "I encountered an error processing your request. Please try again.",
+                    "conversation_id": str(conversation_id) if conversation_id else None
+                }
+            )
+
+
+@router.get("/{user_id}/conversations")
+async def list_conversations(
+    user_id: str,
+    current_user: str = Depends(get_current_user)
+):
+    """
+    List all conversations for a user.
+
+    Returns conversations ordered by most recently updated.
+    """
+    # Verify user ownership
+    if current_user != user_id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Access forbidden: You can only access your own conversations"
+        )
+
+    try:
+        user_uuid = UUID(user_id)
+    except ValueError:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid user ID format"
+        )
+
+    with Session(engine) as session:
+        conversations = ChatService.get_user_conversations(session, user_uuid)
+
+        return {
+            "conversations": [
+                {
+                    "id": str(conv.id),
+                    "title": conv.title,
+                    "created_at": conv.created_at.isoformat(),
+                    "updated_at": conv.updated_at.isoformat()
+                }
+                for conv in conversations
+            ],
+            "count": len(conversations)
+        }
+
+
+@router.get("/{user_id}/conversations/{conversation_id}/messages")
+async def get_conversation_messages(
+    user_id: str,
+    conversation_id: str,
+    current_user: str = Depends(get_current_user)
+):
+    """
+    Get all messages in a conversation.
+
+    Requires user owns the conversation.
+    """
+    # Verify user ownership
+    if current_user != user_id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Access forbidden"
+        )
+
+    try:
+        user_uuid = UUID(user_id)
+        conv_uuid = UUID(conversation_id)
+    except ValueError:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid ID format"
+        )
+
+    with Session(engine) as session:
+        # Verify user owns the conversation
+        conversation = ChatService.get_conversation(session, conv_uuid, user_uuid)
+        if not conversation:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Conversation not found"
+            )
+
+        # Get messages
+        messages = ChatService.get_conversation_history(
+            session=session,
+            conversation_id=conv_uuid,
+            user_id=user_uuid
+        )
+
+        return {
+            "conversation_id": conversation_id,
+            "messages": [
+                {
+                    "id": str(msg.id),
+                    "role": msg.role.value,
+                    "content": msg.content,
+                    "created_at": msg.created_at.isoformat()
+                }
+                for msg in messages
+            ],
+            "count": len(messages)
+        }

api/routes/health.py

@@ -0,0 +1,41 @@
+"""
+Health check endpoint.
+
+Per @specs/001-auth-api-bridge/contracts/pydantic-models.md
+"""
+from fastapi import APIRouter
+from pydantic import BaseModel
+from src.config import settings, engine
+from sqlalchemy import text
+
+
+class HealthResponse(BaseModel):
+    """Health check response."""
+    status: str = "healthy"
+    database: str = "connected"
+    version: str = "1.0.0"
+
+
+router = APIRouter()
+
+
+@router.get("/health", response_model=HealthResponse)
+async def health_check():
+    """
+    Health check endpoint.
+
+    Returns service health status and database connectivity.
+    """
+    # Check database connection
+    try:
+        with engine.connect() as conn:
+            conn.execute(text("SELECT 1"))
+        db_status = "connected"
+    except Exception as e:
+        db_status = f"disconnected: {str(e)[:50]}"
+
+    return HealthResponse(
+        status="healthy" if db_status == "connected" else "unhealthy",
+        database=db_status,
+        version="1.0.0"
+    )

api/routes/tasks.py

@@ -0,0 +1,413 @@
+"""
+Task API routes with JWT authentication and user isolation.
+
+Per @specs/001-auth-api-bridge/api/rest-endpoints.md and
+@specs/001-auth-api-bridge/contracts/pydantic-models.md
+"""
+from fastapi import APIRouter, Depends, HTTPException, status, Request
+from sqlmodel import Session, select
+from typing import List, Optional
+from uuid import UUID
+from datetime import datetime
+
+from src.api.dependencies import get_current_user, verify_user_ownership
+from src.services.task import TaskService
+from src.config import engine
+from src.models.user import UserTable
+from pydantic import BaseModel, Field, constr
+
+
+# =============================================================================
+# Pydantic Models for Request/Response Validation
+# Per @specs/001-auth-api-bridge/contracts/pydantic-models.md
+# =============================================================================
+
+class TaskCreateRequest(BaseModel):
+    """Request model for creating a task."""
+    title: constr(min_length=1, max_length=255, strip_whitespace=True) = Field(
+        ...,
+        description="Task title (1-255 characters)"
+    )
+    description: Optional[constr(max_length=5000)] = Field(
+        None,
+        description="Task description (optional, max 5000 characters)"
+    )
+    priority: str = Field(
+        default="medium",
+        description="Task priority level: low, medium, or high"
+    )
+
+
+class TaskUpdateRequest(BaseModel):
+    """Request model for updating a task."""
+    title: Optional[constr(min_length=1, max_length=255, strip_whitespace=True)] = Field(
+        None,
+        description="Task title (1-255 characters)"
+    )
+    description: Optional[constr(max_length=5000)] = Field(
+        None,
+        description="Task description (optional, max 5000 characters)"
+    )
+    priority: Optional[str] = Field(
+        None,
+        description="Task priority level: low, medium, or high"
+    )
+
+
+class TaskResponse(BaseModel):
+    """Response model for a task."""
+    id: UUID = Field(..., description="Unique task identifier")
+    title: str = Field(..., description="Task title")
+    description: Optional[str] = Field(None, description="Task description")
+    completed: bool = Field(..., description="Task completion status")
+    priority: str = Field(..., description="Task priority level")
+    created_at: str = Field(..., description="Task creation timestamp (ISO 8601)")
+    completed_at: Optional[str] = Field(None, description="Task completion timestamp (ISO 8601)")
+
+    model_config = {"from_attributes": True}
+
+
+class TaskListResponse(BaseModel):
+    """Response model for a list of tasks."""
+    tasks: List[TaskResponse] = Field(..., description="List of tasks")
+    count: int = Field(..., description="Total number of tasks")
+
+
+class ErrorDetail(BaseModel):
+    """Error detail structure."""
+    code: str = Field(..., description="Error code (e.g., UNAUTHORIZED, NOT_FOUND)")
+    message: str = Field(..., description="Human-readable error message")
+    details: dict = Field(default_factory=dict, description="Additional error context")
+
+
+class ErrorResponse(BaseModel):
+    """Standard error response."""
+    error: ErrorDetail
+
+
+# =============================================================================
+# Task Routes with JWT Authentication
+# =============================================================================
+
+router = APIRouter()
+
+
+def ensure_user_exists(session: Session, user_id: UUID) -> None:
+    """Create user if they don't exist in the database."""
+    user = session.get(UserTable, user_id)
+    if user is None:
+        # Create user with a placeholder email
+        user = UserTable(
+            id=user_id,
+            email=f"user-{str(user_id)[:8]}@placeholder.com",
+            created_at=datetime.utcnow(),
+            updated_at=datetime.utcnow()
+        )
+        session.add(user)
+        session.commit()
+        print(f"Created new user: {user_id}")
+
+
+@router.post(
+    "/api/{user_id}/tasks",
+    response_model=TaskResponse,
+    status_code=status.HTTP_201_CREATED,
+    responses={
+        401: {"model": ErrorResponse, "description": "Unauthorized - Invalid or missing token"},
+        403: {"model": ErrorResponse, "description": "Forbidden - User ID mismatch"},
+        400: {"model": ErrorResponse, "description": "Bad Request - Validation error"}
+    }
+)
+async def create_task(
+    user_id: str,
+    task_data: TaskCreateRequest,
+    request: Request,
+    current_user: str = Depends(get_current_user)
+):
+    """
+    Create a new task for the authenticated user.
+
+    Per @specs/001-auth-api-bridge/api/rest-endpoints.md
+
+    Security:
+    - JWT token must be valid and not expired
+    - user_id in path must match JWT sub claim (user ownership)
+    - Task is automatically assigned to authenticated user
+    """
+    # Verify user ownership: user_id in path must match authenticated user
+    await verify_user_ownership(request, user_id)
+
+    # Create task with user_id from verified JWT
+    with Session(engine) as session:
+        # Ensure user exists in database
+        ensure_user_exists(session, UUID(current_user))
+
+        task = TaskService.create_task(
+            session=session,
+            user_id=UUID(current_user),
+            title=task_data.title,
+            description=task_data.description,
+            priority=task_data.priority
+        )
+
+        # Convert datetime objects to ISO 8601 strings for JSON response
+        return TaskResponse(
+            id=task.id,
+            title=task.title,
+            description=task.description,
+            completed=task.completed,
+            priority=task.priority,
+            created_at=task.created_at.isoformat(),
+            completed_at=task.completed_at.isoformat() if task.completed_at else None
+        )
+
+
+@router.get(
+    "/api/{user_id}/tasks",
+    response_model=TaskListResponse,
+    responses={
+        401: {"model": ErrorResponse, "description": "Unauthorized - Invalid or missing token"},
+        403: {"model": ErrorResponse, "description": "Forbidden - User ID mismatch"}
+    }
+)
+async def list_tasks(
+    user_id: str,
+    request: Request,  # type: ignore
+    current_user: str = Depends(get_current_user)
+):
+    """
+    List all tasks for the authenticated user.
+
+    Per @specs/001-auth-api-bridge/api/rest-endpoints.md
+
+    Security:
+    - JWT token must be valid
+    - user_id in path must match JWT sub claim
+    - Only returns tasks owned by authenticated user
+    """
+    await verify_user_ownership(request, user_id)
+
+    with Session(engine) as session:
+        tasks = TaskService.get_user_tasks(session=session, user_id=UUID(current_user))
+
+        return TaskListResponse(
+            tasks=[
+                TaskResponse(
+                    id=task.id,
+                    title=task.title,
+                    description=task.description,
+                    completed=task.completed,
+                    priority=task.priority,
+                    created_at=task.created_at.isoformat(),
+                    completed_at=task.completed_at.isoformat() if task.completed_at else None
+                )
+                for task in tasks
+            ],
+            count=len(tasks)
+        )
+
+
+@router.get(
+    "/api/{user_id}/tasks/{task_id}",
+    response_model=TaskResponse,
+    responses={
+        401: {"model": ErrorResponse, "description": "Unauthorized"},
+        403: {"model": ErrorResponse, "description": "Forbidden - Task belongs to different user"},
+        404: {"model": ErrorResponse, "description": "Task not found"}
+    }
+)
+async def get_task(
+    user_id: str,
+    task_id: str,
+    request: Request,  # type: ignore
+    current_user: str = Depends(get_current_user)
+):
+    """
+    Get details of a specific task.
+
+    Security:
+    - JWT token must be valid
+    - user_id in path must match JWT sub claim
+    - Task must belong to authenticated user
+    """
+    await verify_user_ownership(request, user_id)
+
+    with Session(engine) as session:
+        task = TaskService.get_task_by_id(
+            session=session,
+            task_id=UUID(task_id),
+            user_id=UUID(current_user)
+        )
+
+        if not task:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Task not found"
+            )
+
+        return TaskResponse(
+            id=task.id,
+            title=task.title,
+            description=task.description,
+            completed=task.completed,
+            priority=task.priority,
+            created_at=task.created_at.isoformat(),
+            completed_at=task.completed_at.isoformat() if task.completed_at else None
+        )
+
+
+@router.patch(
+    "/api/{user_id}/tasks/{task_id}/complete",
+    response_model=TaskResponse,
+    responses={
+        401: {"model": ErrorResponse, "description": "Unauthorized"},
+        403: {"model": ErrorResponse, "description": "Forbidden - Task belongs to different user"},
+        404: {"model": ErrorResponse, "description": "Task not found"}
+    }
+)
+async def complete_task(
+    user_id: str,
+    task_id: str,
+    request: Request,  # type: ignore
+    current_user: str = Depends(get_current_user)
+):
+    """
+    Mark a task as completed.
+
+    Per @specs/001-auth-api-bridge/api/rest-endpoints.md
+
+    Security:
+    - JWT token must be valid
+    - user_id in path must match JWT sub claim
+    - Task must belong to authenticated user
+
+    Idempotent: Can be called multiple times with same result
+    """
+    await verify_user_ownership(request, user_id)
+
+    with Session(engine) as session:
+        task = TaskService.complete_task(
+            session=session,
+            task_id=UUID(task_id),
+            user_id=UUID(current_user)
+        )
+
+        if not task:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Task not found"
+            )
+
+        return TaskResponse(
+            id=task.id,
+            title=task.title,
+            description=task.description,
+            completed=task.completed,
+            priority=task.priority,
+            created_at=task.created_at.isoformat(),
+            completed_at=task.completed_at.isoformat() if task.completed_at else None
+        )
+
+
+@router.patch(
+    "/api/{user_id}/tasks/{task_id}",
+    response_model=TaskResponse,
+    responses={
+        401: {"model": ErrorResponse, "description": "Unauthorized"},
+        403: {"model": ErrorResponse, "description": "Forbidden - Task belongs to different user"},
+        404: {"model": ErrorResponse, "description": "Task not found"}
+    }
+)
+async def update_task(
+    user_id: str,
+    task_id: str,
+    task_data: TaskUpdateRequest,
+    request: Request,  # type: ignore
+    current_user: str = Depends(get_current_user)
+):
+    """
+    Update a task's title and/or description.
+
+    Per @specs/001-auth-api-bridge/api/rest-endpoints.md
+
+    Security:
+    - JWT token must be valid
+    - user_id in path must match JWT sub claim
+    - Task must belong to authenticated user
+    """
+    await verify_user_ownership(request, user_id)
+
+    with Session(engine) as session:
+        # Check if at least one field is being updated
+        if task_data.title is None and task_data.description is None and task_data.priority is None:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="At least one field (title, description, or priority) must be provided"
+            )
+
+        task = TaskService.update_task(
+            session=session,
+            task_id=UUID(task_id),
+            user_id=UUID(current_user),
+            title=task_data.title,
+            description=task_data.description,
+            priority=task_data.priority
+        )
+
+        if not task:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Task not found"
+            )
+
+        return TaskResponse(
+            id=task.id,
+            title=task.title,
+            description=task.description,
+            completed=task.completed,
+            priority=task.priority,
+            created_at=task.created_at.isoformat(),
+            completed_at=task.completed_at.isoformat() if task.completed_at else None
+        )
+
+
+
+@router.delete(
+    "/api/{user_id}/tasks/{task_id}",
+    status_code=status.HTTP_204_NO_CONTENT,
+    responses={
+        401: {"model": ErrorResponse, "description": "Unauthorized"},
+        403: {"model": ErrorResponse, "description": "Forbidden - Task belongs to different user"},
+        404: {"model": ErrorResponse, "description": "Task not found"}
+    }
+)
+async def delete_task(
+    user_id: str,
+    task_id: str,
+    request: Request,  # type: ignore
+    current_user: str = Depends(get_current_user)
+):
+    """
+    Delete a task.
+
+    Security:
+    - JWT token must be valid
+    - user_id in path must match JWT sub claim
+    - Task must belong to authenticated user
+    """
+    await verify_user_ownership(request, user_id)
+
+    with Session(engine) as session:
+        success = TaskService.delete_task(
+            session=session,
+            task_id=UUID(task_id),
+            user_id=UUID(current_user)
+        )
+
+        if not success:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Task not found"
+            )
+
+        return None

api/schemas/__init__.py

@@ -0,0 +1,17 @@
+"""Pydantic schemas for API request/response validation."""
+
+from src.api.schemas.chat import (
+    ChatRequest,
+    ChatResponse,
+    Message,
+    TaskSummary,
+    ChatError
+)
+
+__all__ = [
+    "ChatRequest",
+    "ChatResponse",
+    "Message",
+    "TaskSummary",
+    "ChatError"
+]

api/schemas/chat.py

@@ -0,0 +1,151 @@
+"""
+Chat API Pydantic schemas for request/response validation.
+
+Per @specs/001-chatbot-mcp/contracts/openapi.yaml
+"""
+from pydantic import BaseModel, Field
+from typing import Optional, List, Dict, Any
+from datetime import datetime
+from uuid import UUID
+
+
+# =============================================================================
+# Request Schemas
+# =============================================================================
+
+class ChatRequest(BaseModel):
+    """
+    Request schema for chat endpoint.
+
+    Per @specs/001-chatbot-mcp/contracts/openapi.yaml
+    """
+    conversation_id: Optional[UUID] = Field(
+        default=None,
+        description="Optional conversation ID to continue existing chat. If not provided, a new conversation is created."
+    )
+    message: str = Field(
+        ...,
+        min_length=1,
+        max_length=5000,
+        description="User's message content"
+    )
+
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "conversation_id": None,
+                "message": "Add a task to buy groceries"
+            }
+        }
+
+
+# =============================================================================
+# Response Schemas
+# =============================================================================
+
+class Message(BaseModel):
+    """
+    Message schema for chat responses.
+
+    Represents a single message in the conversation.
+    Per @specs/001-chatbot-mcp/data-model.md
+    """
+    id: UUID = Field(..., description="Unique message identifier")
+    role: str = Field(..., description="Message role: 'user' or 'assistant'")
+    content: str = Field(..., description="Message content")
+    created_at: datetime = Field(..., description="Timestamp when message was created")
+
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "id": "123e4567-e89b-12d3-a456-426614174000",
+                "role": "assistant",
+                "content": "I've added a task 'Buy groceries' to your list.",
+                "created_at": "2026-01-11T22:00:00Z"
+            }
+        }
+
+
+class TaskSummary(BaseModel):
+    """
+    Task summary schema for chat responses.
+
+    Simplified task representation returned in chat responses.
+    Per @specs/001-chatbot-mcp/contracts/openapi.yaml
+    """
+    id: UUID = Field(..., description="Task ID")
+    title: str = Field(..., description="Task title")
+    description: Optional[str] = Field(None, description="Task description")
+    completed: bool = Field(..., description="Task completion status")
+
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "id": "123e4567-e89b-12d3-a456-426614174000",
+                "title": "Buy groceries",
+                "description": None,
+                "completed": False
+            }
+        }
+
+
+class ChatResponse(BaseModel):
+    """
+    Response schema for chat endpoint.
+
+    Per @specs/001-chatbot-mcp/contracts/openapi.yaml
+    """
+    conversation_id: UUID = Field(..., description="Conversation ID (new or existing)")
+    message: Message = Field(..., description="Assistant's response message")
+    tasks: Optional[List[TaskSummary]] = Field(
+        default=None,
+        description="List of tasks affected by the chat (optional, for context)"
+    )
+
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "conversation_id": "123e4567-e89b-12d3-a456-426614174000",
+                "message": {
+                    "id": "123e4567-e89b-12d3-a456-426614174001",
+                    "role": "assistant",
+                    "content": "I've added a task 'Buy groceries' to your list.",
+                    "created_at": "2026-01-11T22:00:00Z"
+                },
+                "tasks": [
+                    {
+                        "id": "123e4567-e89b-12d3-a456-426614174002",
+                        "title": "Buy groceries",
+                        "description": None,
+                        "completed": False
+                    }
+                ]
+            }
+        }
+
+
+# =============================================================================
+# Error Schemas
+# =============================================================================
+
+class ChatError(BaseModel):
+    """
+    Error response schema for chat endpoint.
+
+    Returned when chat processing fails.
+    """
+    error: str = Field(..., description="Error message")
+    detail: Optional[str] = Field(None, description="Detailed error information")
+    conversation_id: Optional[UUID] = Field(
+        None,
+        description="Conversation ID if error occurred during existing conversation"
+    )
+
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "error": "Failed to process chat message",
+                "detail": "OpenAI API timeout",
+                "conversation_id": None
+            }
+        }

config.py

@@ -0,0 +1,81 @@
+"""
+Backend configuration with environment variable loading.
+
+Per @specs/001-auth-api-bridge/research.md
+"""
+import os
+from functools import lru_cache
+from pathlib import Path
+from dotenv import load_dotenv
+from pydantic_settings import BaseSettings
+from sqlalchemy import create_engine
+from sqlmodel import SQLModel
+
+# Load .env file with override to ensure .env takes precedence over system env vars
+# This is needed when system env vars are set with placeholder values
+env_path = Path(__file__).parent.parent / ".env"
+load_dotenv(env_path, override=True)
+
+
+class Settings(BaseSettings):
+    """Application settings loaded from environment variables."""
+
+    # Better Auth
+    better_auth_secret: str
+    better_auth_url: str = "http://localhost:3000"
+
+    # Database
+    database_url: str
+
+    # API
+    api_port: int = 8000
+    api_host: str = "localhost"
+    debug: bool = True
+
+    # Chatbot Configuration
+    # Per @specs/001-chatbot-mcp/plan.md
+    openai_api_key: str
+    neon_database_url: str  # Same as database_url but explicit for chatbot
+    mcp_server_port: int = 8000
+    openai_model: str = "gpt-4-turbo-preview"
+    mcp_server_host: str = "127.0.0.1"
+
+    # Email Configuration
+    email_host: str = "smtp.gmail.com"
+    email_port: int = 587
+    email_username: str = ""
+    email_password: str = ""
+    email_from: str = ""
+    email_from_name: str = "TaskFlow"
+    emails_enabled: bool = True
+
+    class Config:
+        env_file = ".env"
+        case_sensitive = False
+        extra = "ignore"  # Ignore extra env vars (frontend vars)
+
+
+@lru_cache()
+def get_settings() -> Settings:
+    """Get cached settings instance."""
+    return Settings()
+
+
+# Global settings instance
+settings = get_settings()
+
+# Database engine with connection pooling per @specs/001-auth-api-bridge/research.md
+engine = create_engine(
+    settings.database_url,
+    poolclass=None,  # QueuePool (default)
+    pool_size=5,  # Connections to maintain
+    max_overflow=10,  # Additional connections under load
+    pool_pre_ping=True,  # Validate connections before use (handles Neon scale-to-zero)
+    pool_recycle=3600,  # Recycle connections after 1 hour
+    echo=settings.debug,  # Log SQL in development
+)
+
+
+def init_db():
+    """Initialize database tables."""
+    SQLModel.metadata.create_all(engine)

dockerfile

@@ -0,0 +1,12 @@
+FROM python:3.11-slim
+
+WORKDIR /
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+EXPOSE 7860
+
+CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "7860"]

main.py

@@ -0,0 +1,240 @@
+"""
+FastAPI application entry point.
+
+Per @specs/001-auth-api-bridge/plan.md and @specs/001-auth-api-bridge/quickstart.md
+
+Includes password reset functionality.
+"""
+from datetime import datetime, timedelta
+from uuid import uuid4, UUID
+
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+
+from src.config import settings
+from src.api.routes import tasks_router, health_router, chat_router
+from src.services.auth import create_token, create_password_reset_token, verify_password_reset_token, consume_password_reset_token
+from src.services.email import send_password_reset_email
+
+
+# Create FastAPI application
+app = FastAPI(
+    title="Task Management API",
+    description="FastAPI backend for task management with JWT authentication",
+    version="1.0.0",
+    docs_url="/docs",
+    redoc_url="/redoc"
+)
+
+# =============================================================================
+# CORS Middleware
+# Per @specs/001-auth-api-bridge/quickstart.md
+# =============================================================================
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=[
+        "http://localhost:3000",
+        "http://localhost:3001",
+        "http://localhost:3002",
+        "http://127.0.0.1:3000",
+        "http://127.0.0.1:3001",
+        "http://127.0.0.1:3002",
+        "https://taskflow-app-frontend-4kmp-emsultiio-mawbs-projects.vercel.app",
+    ],  # Frontend URLs
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# =============================================================================
+# Register Routes
+# =============================================================================
+app.include_router(health_router)
+app.include_router(tasks_router)
+app.include_router(chat_router)
+
+
+# =============================================================================
+# Startup Event
+# =============================================================================
+@app.on_event("startup")
+async def startup_event():
+    """
+    Initialize database tables and MCP server on startup.
+
+    Per @specs/001-chatbot-mcp/plan.md, MCP server lifecycle is tied to FastAPI app.
+    """
+    from src.config import init_db
+    from src.services.mcp import mcp_service
+    import logging
+
+    logger = logging.getLogger(__name__)
+
+    # Initialize database tables
+    try:
+        init_db()
+        logger.info("Database initialized successfully")
+    except Exception as e:
+        logger.error(f"Database initialization failed: {e}")
+        # Don't fail startup - database might be available later
+
+    # Initialize MCP server with all tools registered
+    try:
+        await mcp_service.initialize()
+        logger.info("MCP server initialized successfully")
+    except Exception as e:
+        logger.error(f"MCP server initialization failed: {e}")
+        # Don't fail startup - MCP might not be critical for basic functionality
+
+
+@app.on_event("shutdown")
+async def shutdown_event():
+    """
+    Shutdown MCP server gracefully.
+
+    Per @specs/001-chatbot-mcp/plan.md, MCP server lifecycle is tied to FastAPI app.
+    """
+    from src.services.mcp import mcp_service
+    await mcp_service.shutdown()
+
+
+@app.get("/")
+async def root():
+    """Root endpoint."""
+    return {
+        "message": "Task Management API",
+        "version": "1.0.0",
+        "docs": "/docs"
+    }
+
+
+# =============================================================================
+# Demo Token Generation (for testing login page)
+# =============================================================================
+class TokenRequest(BaseModel):
+    email: str
+
+
+@app.post("/generate-token")
+async def generate_token(request: TokenRequest):
+    """
+    Generate a test JWT token for demo purposes.
+
+    In production, this would be replaced by Better Auth's actual authentication.
+
+    Uses consistent user_id generation based on email hash to ensure
+    users always get the same user_id when logging in with the same email.
+    This fixes the issue where tasks appeared "lost" after logout/login.
+    """
+    import hashlib
+    import uuid
+
+    # Generate a consistent user ID based on email hash
+    # This ensures the same email always gets the same user_id
+    email_bytes = request.email.encode('utf-8')
+    hash_bytes = hashlib.sha256(email_bytes).digest()
+    # Convert first 16 bytes to a UUID (UUID v5 style but using SHA256)
+    user_id = str(uuid.UUID(bytes=hash_bytes[:16]))
+
+    # Ensure user exists in database (create if not)
+    from sqlmodel import Session
+    from src.models.user import UserTable
+    from src.config import engine
+
+    with Session(engine) as session:
+        existing_user = session.get(UserTable, user_id)
+        if not existing_user:
+            # Create new user with this email
+            from datetime import datetime
+            user = UserTable(
+                id=user_id,
+                email=request.email,
+                created_at=datetime.utcnow(),
+                updated_at=datetime.utcnow()
+            )
+            session.add(user)
+            session.commit()
+            print(f"Created new user: {user_id} with email: {request.email}")
+
+    # Create JWT token
+    token = create_token(user_id)
+
+    return {
+        "token": token,
+        "userId": user_id,
+        "email": request.email
+    }
+
+
+# =============================================================================
+# Password Reset (Demo Mode)
+# =============================================================================
+class ForgotPasswordRequest(BaseModel):
+    email: str
+
+
+class ResetPasswordRequest(BaseModel):
+    token: str
+    new_password: str
+
+
+@app.post("/forgot-password")
+async def forgot_password(request: ForgotPasswordRequest):
+    """
+    Initiate password reset process.
+
+    Sends an email with a password reset link.
+    In demo mode (without email configured), the token is logged and can be used directly.
+    """
+    # Create a password reset token
+    reset_token = create_password_reset_token(request.email)
+
+    # Send password reset email
+    email_sent = await send_password_reset_email(
+        email=request.email,
+        reset_token=reset_token,
+        frontend_url="http://localhost:3002"  # In production, use settings.better_auth_url
+    )
+
+    # For demo mode, include token in response if email wasn't actually sent
+    # In production with real email, never include the token in the response
+    include_token = not settings.emails_enabled or not settings.email_username
+
+    response_data = {
+        "message": "If an account exists with that email, a password reset link has been sent.",
+        "email": request.email
+    }
+
+    if include_token:
+        response_data["reset_token"] = reset_token
+        response_data["demo_mode"] = True
+
+    return response_data
+
+
+@app.post("/reset-password")
+async def reset_password(request: ResetPasswordRequest):
+    """
+    Reset password using the token received from forgot-password.
+
+    In production, this would update the user's password in the database.
+    For demo purposes, it just validates the token.
+    """
+    # Verify the reset token
+    email = verify_password_reset_token(request.token)
+
+    if email is None:
+        raise HTTPException(
+            status_code=400,
+            detail="Invalid or expired reset token"
+        )
+
+    # In production, you would update the password in the database here
+    # For demo, we just consume the token
+    consume_password_reset_token(request.token)
+
+    return {
+        "message": "Password reset successfully",
+        "email": email
+    }

mcp/__init__.py

@@ -0,0 +1,5 @@
+"""MCP server and tools for Todo AI Chatbot integration."""
+
+from src.mcp.server import get_mcp_server, mcp_server
+
+__all__ = ["get_mcp_server", "mcp_server"]

mcp/server.py

@@ -0,0 +1,117 @@
+"""
+MCP Server initialization for Todo task management.
+
+Per @specs/001-chatbot-mcp/plan.md Section VIII - AI Chatbot Architecture
+MCP First: All task operations go through MCP SDK for OpenAI Agents integration.
+
+Note: We create a simple tool registry instead of using FastMCP server
+to avoid transport initialization issues in embedded mode.
+"""
+from typing import List, Dict, Any, Callable, Optional, Union
+from dataclasses import dataclass
+
+
+@dataclass
+class Tool:
+    """Simple tool definition for task management."""
+    name: str
+    description: str
+    parameters: Dict[str, Any]
+    handler: Callable
+
+
+class SimpleMCPRegistry:
+    """Simple tool registry for MCP-compatible tools without server overhead."""
+
+    def __init__(self, name: str, instructions: str):
+        self.name = name
+        self.instructions = instructions
+        self._tools: Dict[str, Tool] = {}
+
+    def tool(self, name: Optional[str] = None, description: Optional[str] = None):
+        """Decorator to register tools."""
+        def decorator(func: Callable):
+            tool_name = name or func.__name__
+            self._tools[tool_name] = Tool(
+                name=tool_name,
+                description=description or func.__doc__ or "",
+                parameters=self._get_parameters_from_func(func),
+                handler=func
+            )
+            return func
+        return decorator
+
+    def _get_parameters_from_func(self, func: Callable) -> Dict[str, Any]:
+        """Extract parameters from function signature."""
+        import inspect
+        sig = inspect.signature(func)
+        properties = {}
+        required = []
+
+        for param_name, param in sig.parameters.items():
+            param_type = param.annotation if param.annotation != inspect.Parameter.empty else "string"
+            properties[param_name] = {
+                "type": self._get_type_string(param_type),
+                "description": f"{param_name} parameter"
+            }
+            if param.default == inspect.Parameter.empty:
+                required.append(param_name)
+
+        return {
+            "type": "object",
+            "properties": properties,
+            "required": required
+        }
+
+    def _get_type_string(self, type_hint) -> str:
+        """Convert type hint to JSON schema type string."""
+        type_map = {
+            str: "string",
+            int: "integer",
+            float: "number",
+            bool: "boolean",
+            list: "array",
+            dict: "object"
+        }
+        if type_hint in type_map:
+            return type_map[type_hint]
+        # Handle Optional types and other generics
+        if hasattr(type_hint, "__origin__"):
+            origin = getattr(type_hint, "__origin__", None)
+            if origin is Union:
+                return "string"
+            if origin is list:
+                return "array"
+        return "string"
+
+    def list_tools(self) -> List[Tool]:
+        """List all registered tools."""
+        return list(self._tools.values())
+
+    def get_tool(self, name: str) -> Optional[Tool]:
+        """Get a tool by name."""
+        return self._tools.get(name)
+
+
+# Create the tool registry
+mcp_server = SimpleMCPRegistry(
+    name="todo-mcp-server",
+    instructions="MCP server for Todo task management operations. Provides tools for creating, listing, completing, deleting, and updating tasks with user isolation."
+)
+
+
+def get_mcp_server() -> SimpleMCPRegistry:
+    """
+    Get the MCP server/tool registry instance.
+
+    Returns:
+        The configured tool registry with all tools registered.
+
+    This function is called by the FastAPI application during startup
+    to initialize the MCP server lifecycle.
+    """
+    # Import and register tools
+    from src.mcp.tools import register_task_tools
+    register_task_tools(mcp_server)
+
+    return mcp_server

mcp/tools.py

@@ -0,0 +1,314 @@
+"""
+MCP tool definitions for Todo task management.
+
+Per @specs/001-chatbot-mcp/contracts/mcp-tools.json
+All tools enforce user isolation by requiring user_id in every request.
+"""
+from sqlmodel import Session, select
+from typing import Optional, TYPE_CHECKING
+from datetime import datetime
+from uuid import UUID
+
+from src.models.task import TaskTable
+from src.config import engine
+
+if TYPE_CHECKING:
+    from src.mcp.server import SimpleMCPRegistry
+
+
+def register_task_tools(mcp_server) -> None:
+    """
+    Register all task management tools with the MCP server.
+
+    Args:
+        mcp_server: The SimpleMCPRegistry instance
+
+    This function is called during MCP server initialization to register
+    all 5 task tools: add_task, list_tasks, complete_task, delete_task, update_task.
+    Per @specs/001-chatbot-mcp/plan.md, all task operations go through MCP.
+    """
+
+    @mcp_server.tool()
+    async def add_task(
+        user_id: str,
+        title: str,
+        description: Optional[str] = None
+    ) -> dict:
+        """
+        Create a new todo task for a user.
+
+        Use this when the user requests to create, add, or make a new task.
+
+        Args:
+            user_id: ID of the user who will own the task (UUID string)
+            title: Task title (short, descriptive name)
+            description: Optional detailed description
+
+        Returns:
+            Dict with task_id, title, description, created_at, success, error
+        """
+        try:
+            with Session(engine) as session:
+                task = TaskTable(
+                    user_id=UUID(user_id),
+                    title=title,
+                    description=description,
+                    completed=False,
+                    created_at=datetime.utcnow()
+                )
+                session.add(task)
+                session.commit()
+                session.refresh(task)
+
+                return {
+                    "task_id": str(task.id),
+                    "title": task.title,
+                    "description": task.description,
+                    "created_at": task.created_at.isoformat(),
+                    "success": True,
+                    "error": None
+                }
+        except Exception as e:
+            return {
+                "task_id": None,
+                "title": None,
+                "description": None,
+                "created_at": None,
+                "success": False,
+                "error": str(e)
+            }
+
+    @mcp_server.tool()
+    async def list_tasks(
+        user_id: str,
+        include_completed: bool = True
+    ) -> dict:
+        """
+        List all tasks for a user.
+
+        Use this when the user asks to see, show, or display their tasks.
+
+        Args:
+            user_id: ID of the user whose tasks to list (UUID string)
+            include_completed: Whether to include completed tasks (default: true)
+
+        Returns:
+            Dict with tasks list, count, success, error
+        """
+        try:
+            with Session(engine) as session:
+                statement = select(TaskTable).where(TaskTable.user_id == UUID(user_id))
+
+                if not include_completed:
+                    statement = statement.where(TaskTable.completed == False)
+
+                tasks = session.exec(statement).all()
+
+                return {
+                    "tasks": [
+                        {
+                            "id": str(task.id),
+                            "title": task.title,
+                            "description": task.description,
+                            "completed": task.completed,
+                            "created_at": task.created_at.isoformat(),
+                            "completed_at": task.completed_at.isoformat() if task.completed_at else None
+                        }
+                        for task in tasks
+                    ],
+                    "count": len(tasks),
+                    "success": True,
+                    "error": None
+                }
+        except Exception as e:
+            return {
+                "tasks": [],
+                "count": 0,
+                "success": False,
+                "error": str(e)
+            }
+
+    @mcp_server.tool()
+    async def complete_task(
+        user_id: str,
+        task_id: str
+    ) -> dict:
+        """
+        Mark a task as completed.
+
+        Use this when the user asks to complete, finish, or check off a task.
+
+        Args:
+            user_id: ID of the user who owns the task (UUID string)
+            task_id: ID of the task to mark as complete (UUID string)
+
+        Returns:
+            Dict with task_id, title, completed, completed_at, success, error
+        """
+        try:
+            with Session(engine) as session:
+                task = session.query(TaskTable).filter(
+                    TaskTable.id == UUID(task_id),
+                    TaskTable.user_id == UUID(user_id)
+                ).first()
+
+                if not task:
+                    return {
+                        "task_id": task_id,
+                        "title": None,
+                        "completed": False,
+                        "completed_at": None,
+                        "success": False,
+                        "error": "Task not found or access denied"
+                    }
+
+                task.completed = True
+                task.completed_at = datetime.utcnow()
+                session.add(task)
+                session.commit()
+                session.refresh(task)
+
+                return {
+                    "task_id": str(task.id),
+                    "title": task.title,
+                    "completed": task.completed,
+                    "completed_at": task.completed_at.isoformat(),
+                    "success": True,
+                    "error": None
+                }
+        except Exception as e:
+            return {
+                "task_id": task_id,
+                "title": None,
+                "completed": False,
+                "completed_at": None,
+                "success": False,
+                "error": str(e)
+            }
+
+    @mcp_server.tool()
+    async def delete_task(
+        user_id: str,
+        task_id: str
+    ) -> dict:
+        """
+        Delete a task permanently.
+
+        CRITICAL: The task_id parameter must be a valid UUID from the user's existing tasks.
+        If the user provides a task number or title, you MUST map it to the correct task_id.
+        Example: If user says "delete task 1", find the task with index 0 in their task list and use its ID.
+        Always confirm which task you're deleting by showing the task title before proceeding.
+
+        Use this when the user asks to delete, remove, or get rid of a task.
+
+        Args:
+            user_id: ID of the user who owns the task (UUID string)
+            task_id: ID of the task to delete (UUID string) - must match an existing task ID for this user
+
+        Returns:
+            Dict with task_id, title, deleted, success, error
+        """
+        try:
+            with Session(engine) as session:
+                task = session.query(TaskTable).filter(
+                    TaskTable.id == UUID(task_id),
+                    TaskTable.user_id == UUID(user_id)
+                ).first()
+
+                if not task:
+                    return {
+                        "task_id": task_id,
+                        "title": None,
+                        "deleted": False,
+                        "success": False,
+                        "error": "Task not found or access denied"
+                    }
+
+                title = task.title
+                session.delete(task)
+                session.commit()
+
+                return {
+                    "task_id": task_id,
+                    "title": title,
+                    "deleted": True,
+                    "success": True,
+                    "error": None
+                }
+        except Exception as e:
+            return {
+                "task_id": task_id,
+                "title": None,
+                "deleted": False,
+                "success": False,
+                "error": str(e)
+            }
+
+    @mcp_server.tool()
+    async def update_task(
+        user_id: str,
+        task_id: str,
+        title: Optional[str] = None,
+        description: Optional[str] = None
+    ) -> dict:
+        """
+        Update a task's title or description.
+
+        CRITICAL: The task_id parameter must be a valid UUID from the user's existing tasks.
+        If the user provides a task number or title, you MUST map it to the correct task_id.
+        Example: If user says "update task 1 to buy milk", find the task with index 0 in their task list and use its ID.
+
+        Use this when the user asks to change, modify, or edit a task.
+
+        Args:
+            user_id: ID of the user who owns the task (UUID string)
+            task_id: ID of the task to update (UUID string) - must match an existing task ID for this user
+            title: New task title (optional)
+            description: New task description (optional, empty string to clear)
+
+        Returns:
+            Dict with task_id, title, description, updated_at, success, error
+        """
+        try:
+            with Session(engine) as session:
+                task = session.query(TaskTable).filter(
+                    TaskTable.id == UUID(task_id),
+                    TaskTable.user_id == UUID(user_id)
+                ).first()
+
+                if not task:
+                    return {
+                        "task_id": task_id,
+                        "title": None,
+                        "description": None,
+                        "updated_at": None,
+                        "success": False,
+                        "error": "Task not found or access denied"
+                    }
+
+                if title is not None:
+                    task.title = title
+                if description is not None:
+                    task.description = description if description else None
+
+                session.add(task)
+                session.commit()
+                session.refresh(task)
+
+                return {
+                    "task_id": str(task.id),
+                    "title": task.title,
+                    "description": task.description,
+                    "updated_at": task.created_at.isoformat(),  # Using created_at as updated_at not in model
+                    "success": True,
+                    "error": None
+                }
+        except Exception as e:
+            return {
+                "task_id": task_id,
+                "title": None,
+                "description": None,
+                "updated_at": None,
+                "success": False,
+                "error": str(e)
+            }

models/__init__.py

@@ -0,0 +1,7 @@
+"""Database models package."""
+from src.models.user import UserTable
+from src.models.task import TaskTable
+from src.models.conversation import ConversationTable
+from src.models.message import MessageTable, MessageRole
+
+__all__ = ["UserTable", "TaskTable", "ConversationTable", "MessageTable", "MessageRole"]

models/conversation.py

@@ -0,0 +1,68 @@
+"""
+Conversation model representing a user's chat session with AI.
+
+Per @specs/001-chatbot-mcp/data-model.md
+"""
+from sqlmodel import SQLModel, Field, Relationship
+from typing import TYPE_CHECKING, Optional
+from datetime import datetime
+from uuid import UUID, uuid4
+
+if TYPE_CHECKING:
+    from src.models.user import UserTable
+    from src.models.message import MessageTable
+
+
+class ConversationTable(SQLModel, table=True):
+    """
+    A chat session between a user and the AI assistant.
+
+    All conversations MUST be scoped to a single user to ensure data isolation
+    per constitutional principle III (User Isolation via JWT).
+    """
+    __tablename__ = "conversations"
+
+    # Primary key
+    id: UUID = Field(
+        default_factory=uuid4,
+        primary_key=True,
+        index=True,
+        description="Unique conversation identifier"
+    )
+
+    # Foreign key to User
+    user_id: UUID = Field(
+        foreign_key="users.id",
+        index=True,
+        nullable=False,
+        description="ID of the user who owns this conversation"
+    )
+
+    # Conversation attributes
+    title: Optional[str] = Field(
+        default=None,
+        max_length=255,
+        description="Auto-generated title from first message (e.g., 'Grocery shopping')"
+    )
+
+    # Timestamps
+    created_at: datetime = Field(
+        default_factory=datetime.utcnow,
+        nullable=False,
+        description="Timestamp when conversation was created"
+    )
+
+    updated_at: datetime = Field(
+        default_factory=datetime.utcnow,
+        sa_column_kwargs={"onupdate": datetime.utcnow},
+        nullable=False,
+        index=True,
+        description="Timestamp of last message in conversation"
+    )
+
+    # Relationships
+    user: "UserTable" = Relationship(back_populates="conversations")
+    messages: list["MessageTable"] = Relationship(
+        back_populates="conversation",
+        sa_relationship_kwargs={"cascade": "all, delete-orphan"}
+    )

models/message.py

@@ -0,0 +1,77 @@
+"""
+Message model representing a single message in a conversation.
+
+Per @specs/001-chatbot-mcp/data-model.md
+"""
+from sqlmodel import SQLModel, Field, Relationship, Column
+from typing import TYPE_CHECKING, Optional, Any, Dict
+from datetime import datetime
+from uuid import UUID, uuid4
+from enum import Enum
+from sqlalchemy import JSON
+
+if TYPE_CHECKING:
+    from src.models.conversation import ConversationTable
+
+
+class MessageRole(str, Enum):
+    """Message sender role."""
+    USER = "user"
+    ASSISTANT = "assistant"
+
+
+class MessageTable(SQLModel, table=True):
+    """
+    A single message in a conversation.
+
+    Messages are owned by a user through their conversation. All queries MUST
+    filter by user_id (via conversation) to ensure data isolation.
+    """
+    __tablename__ = "messages"
+
+    # Primary key
+    id: UUID = Field(
+        default_factory=uuid4,
+        primary_key=True,
+        index=True,
+        description="Unique message identifier"
+    )
+
+    # Foreign key to Conversation
+    conversation_id: UUID = Field(
+        foreign_key="conversations.id",
+        index=True,
+        nullable=False,
+        description="ID of the conversation this message belongs to"
+    )
+
+    # Message attributes
+    role: MessageRole = Field(
+        nullable=False,
+        description="Message sender: 'user' or 'assistant'"
+    )
+
+    content: str = Field(
+        nullable=False,
+        max_length=5000,
+        description="Message content (plaintext)"
+    )
+
+    # Timestamps
+    created_at: datetime = Field(
+        default_factory=datetime.utcnow,
+        nullable=False,
+        index=True,
+        description="Timestamp when message was created"
+    )
+
+    # Optional metadata for tool calls, token usage, etc.
+    # Renamed from 'metadata' to avoid conflict with SQLAlchemy's reserved attribute
+    tool_metadata: Optional[Dict[str, Any]] = Field(
+        default=None,
+        sa_column=Column(JSON),
+        description="Tool calls, tokens used, error information"
+    )
+
+    # Relationships
+    conversation: "ConversationTable" = Relationship(back_populates="messages")

models/task.py

@@ -0,0 +1,82 @@
+"""
+Task model representing user tasks.
+
+Per @specs/001-auth-api-bridge/data-model.md
+"""
+from sqlmodel import SQLModel, Field, Relationship
+from typing import TYPE_CHECKING, Optional
+from datetime import datetime
+from uuid import UUID, uuid4
+
+if TYPE_CHECKING:
+    from src.models.user import UserTable
+
+
+class TaskTable(SQLModel, table=True):
+    """
+    Task owned by a user.
+
+    Each task belongs to exactly one user. All queries MUST filter by user_id
+    to ensure data isolation per constitutional principle.
+    """
+    __tablename__ = "tasks"
+
+    # Primary key
+    id: UUID = Field(
+        default_factory=uuid4,
+        primary_key=True,
+        index=True,
+        description="Unique task identifier"
+    )
+
+    # Foreign key to User
+    user_id: UUID = Field(
+        foreign_key="users.id",
+        index=True,
+        nullable=False,
+        description="ID of the user who owns this task"
+    )
+
+    # Task attributes
+    title: str = Field(
+        max_length=255,
+        nullable=False,
+        description="Short title of the task"
+    )
+
+    description: Optional[str] = Field(
+        default=None,
+        max_length=5000,
+        description="Detailed description of the task (optional)"
+    )
+
+    # Completion status
+    completed: bool = Field(
+        default=False,
+        index=True,
+        description="Whether the task has been completed"
+    )
+
+    # Priority level
+    priority: str = Field(
+        default="medium",
+        nullable=False,
+        description="Task priority level: low, medium, or high"
+    )
+
+    # Timestamps
+    created_at: datetime = Field(
+        default_factory=datetime.utcnow,
+        nullable=False,
+        description="Timestamp when task was created"
+    )
+
+    completed_at: Optional[datetime] = Field(
+        default=None,
+        nullable=True,
+        index=True,
+        description="Timestamp when task was marked as completed (null until completed)"
+    )
+
+    # Relationships
+    user: "UserTable" = Relationship(back_populates="tasks")

models/user.py

@@ -0,0 +1,66 @@
+"""
+User model representing authenticated users managed by Better Auth.
+
+Per @specs/001-auth-api-bridge/data-model.md
+"""
+from sqlmodel import SQLModel, Field, Relationship
+from typing import TYPE_CHECKING, List
+from datetime import datetime
+from uuid import UUID, uuid4
+
+if TYPE_CHECKING:
+    from src.models.task import TaskTable
+    from src.models.conversation import ConversationTable
+
+
+class UserTable(SQLModel, table=True):
+    """
+    User account managed by Better Auth.
+
+    The id field (UUID) MUST match the 'sub' claim in JWT tokens issued by Better Auth.
+    """
+    __tablename__ = "users"
+
+    # Primary key - matches the 'sub' claim in JWT tokens
+    id: UUID = Field(
+        default_factory=uuid4,
+        primary_key=True,
+        index=True,
+        description="Unique user identifier that matches JWT 'sub' claim"
+    )
+
+    # User profile information
+    email: str = Field(
+        unique=True,
+        index=True,
+        max_length=255,
+        description="User's email address (unique)"
+    )
+
+    name: str | None = Field(
+        default=None,
+        max_length=255,
+        description="User's display name"
+    )
+
+    # Timestamps
+    created_at: datetime = Field(
+        default_factory=datetime.utcnow,
+        description="Timestamp when user account was created"
+    )
+
+    updated_at: datetime = Field(
+        default_factory=datetime.utcnow,
+        sa_column_kwargs={"onupdate": datetime.utcnow},
+        description="Timestamp when user record was last updated"
+    )
+
+    # Relationships
+    tasks: List["TaskTable"] = Relationship(
+        back_populates="user",
+        sa_relationship_kwargs={"cascade": "all, delete-orphan"}
+    )
+    conversations: List["ConversationTable"] = Relationship(
+        back_populates="user",
+        sa_relationship_kwargs={"cascade": "all, delete-orphan"}
+    )

requirements.txt

@@ -0,0 +1,9 @@
+fastapi
+uvicorn
+pydantic
+python-dotenv
+sqlalchemy
+openai
+python-jose[cryptography]
+passlib[bcrypt]
+python-multipart

services/__init__.py

@@ -0,0 +1,5 @@
+"""Services package."""
+from src.services.auth import verify_token
+from src.services.task import TaskService
+
+__all__ = ["verify_token", "TaskService"]

services/auth.py

@@ -0,0 +1,121 @@
+"""
+JWT verification service for Better Auth integration.
+
+Per @specs/001-auth-api-bridge/research.md
+"""
+from datetime import datetime, timedelta
+from typing import Optional, Dict
+from jose import JWTError, jwt
+from src.config import settings
+import uuid
+
+SECRET_KEY = settings.better_auth_secret
+ALGORITHM = "HS256"
+
+# In-memory store for password reset tokens (in production, use Redis or database)
+PASSWORD_RESET_TOKENS: Dict[str, dict] = {}
+
+
+def create_token(user_id: str, expires_delta: Optional[timedelta] = None) -> str:
+    """
+    Create a JWT token for a user.
+
+    Args:
+        user_id: User UUID to embed in the token
+        expires_delta: Optional custom expiration time
+
+    Returns:
+        Encoded JWT token string
+    """
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(hours=24)
+
+    payload = {
+        "sub": user_id,
+        "iat": datetime.utcnow(),
+        "exp": expire
+    }
+
+    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
+
+
+async def verify_token(token: str) -> Optional[str]:
+    """
+    Verify JWT token and return user_id.
+
+    Args:
+        token: JWT token string
+
+    Returns:
+        User ID (UUID string) if token is valid, None otherwise
+    """
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        user_id: str = payload.get("sub")
+        if user_id is None:
+            return None
+        return user_id
+    except JWTError:
+        return None
+
+
+def create_password_reset_token(email: str) -> str:
+    """
+    Create a password reset token for an email.
+
+    Args:
+        email: User's email address
+
+    Returns:
+        Password reset token
+    """
+    reset_token = str(uuid.uuid4())
+    expiry = datetime.utcnow() + timedelta(hours=1)  # Token valid for 1 hour
+
+    PASSWORD_RESET_TOKENS[reset_token] = {
+        "email": email,
+        "expires": expiry
+    }
+
+    return reset_token
+
+
+def verify_password_reset_token(token: str) -> Optional[str]:
+    """
+    Verify password reset token and return email.
+
+    Args:
+        token: Password reset token
+
+    Returns:
+        Email if token is valid, None otherwise
+    """
+    if token not in PASSWORD_RESET_TOKENS:
+        return None
+
+    token_data = PASSWORD_RESET_TOKENS[token]
+
+    # Check if token has expired
+    if datetime.utcnow() > token_data["expires"]:
+        del PASSWORD_RESET_TOKENS[token]
+        return None
+
+    return token_data["email"]
+
+
+def consume_password_reset_token(token: str) -> bool:
+    """
+    Consume (invalidate) a password reset token after use.
+
+    Args:
+        token: Password reset token to consume
+
+    Returns:
+        True if token was valid and consumed, False otherwise
+    """
+    if token in PASSWORD_RESET_TOKENS:
+        del PASSWORD_RESET_TOKENS[token]
+        return True
+    return False