diff --git a/CLAUDE.md b/CLAUDE.md
index 6517e617..a9fac97a 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -157,8 +157,6 @@ import logging
 log = logging.getLogger(LOGGER_NAME("my_service"))
 ```
 
-New logger names must be added to `alembic.ini` under the logger section.
-
 ## Testing
 
 - Backend tests: `pdm run tests` (runs pytest in `backend/tests/`)
diff --git a/backend/app/interfaces/task_service.py b/backend/app/interfaces/task_service.py
new file mode 100644
index 00000000..d5fc881f
--- /dev/null
+++ b/backend/app/interfaces/task_service.py
@@ -0,0 +1,116 @@
+from abc import ABC, abstractmethod
+from typing import List, Optional
+from uuid import UUID
+
+from app.schemas.task import TaskCreateRequest, TaskResponse, TaskUpdateRequest
+
+
+class ITaskService(ABC):
+    """
+    TaskService interface with task management methods
+    """
+
+    @abstractmethod
+    async def create_task(self, task: TaskCreateRequest) -> TaskResponse:
+        """
+        Create a new task
+
+        :param task: the task to be created
+        :type task: TaskCreateRequest
+        :return: the created task
+        :rtype: TaskResponse
+        :raises Exception: if task creation fails
+        """
+        pass
+
+    @abstractmethod
+    async def get_task_by_id(self, task_id: UUID) -> TaskResponse:
+        """
+        Get task associated with task_id
+
+        :param task_id: task's id
+        :type task_id: UUID
+        :return: a TaskResponse with task's information
+        :rtype: TaskResponse
+        :raises Exception: if task retrieval fails
+        """
+        pass
+
+    @abstractmethod
+    async def get_all_tasks(
+        self,
+        status: Optional[str] = None,
+        priority: Optional[str] = None,
+        task_type: Optional[str] = None,
+        assignee_id: Optional[UUID] = None,
+    ) -> List[TaskResponse]:
+        """
+        Get all tasks with optional filters
+
+        :param status: filter by task status
+        :type status: str, optional
+        :param priority: filter by task priority
+        :type priority: str, optional
+        :param task_type: filter by task type
+        :type task_type: str, optional
+        :param assignee_id: filter by assignee
+        :type assignee_id: UUID, optional
+        :return: list of TaskResponse
+        :rtype: List[TaskResponse]
+        :raises Exception: if task retrieval fails
+        """
+        pass
+
+    @abstractmethod
+    async def update_task(self, task_id: UUID, task_update: TaskUpdateRequest) -> TaskResponse:
+        """
+        Update a task
+
+        :param task_id: task's id
+        :type task_id: UUID
+        :param task_update: the task updates
+        :type task_update: TaskUpdateRequest
+        :return: the updated task
+        :rtype: TaskResponse
+        :raises Exception: if task update fails
+        """
+        pass
+
+    @abstractmethod
+    async def assign_task(self, task_id: UUID, assignee_id: UUID) -> TaskResponse:
+        """
+        Assign a task to an admin user
+
+        :param task_id: task's id
+        :type task_id: UUID
+        :param assignee_id: admin user's id to assign
+        :type assignee_id: UUID
+        :return: the updated task
+        :rtype: TaskResponse
+        :raises Exception: if task assignment fails
+        """
+        pass
+
+    @abstractmethod
+    async def complete_task(self, task_id: UUID) -> TaskResponse:
+        """
+        Mark a task as completed
+
+        :param task_id: task's id
+        :type task_id: UUID
+        :return: the completed task
+        :rtype: TaskResponse
+        :raises Exception: if task completion fails
+        """
+        pass
+
+    @abstractmethod
+    async def delete_task(self, task_id: UUID) -> None:
+        """
+        Delete a task by task_id
+
+        :param task_id: task_id of task to be deleted
+        :type task_id: UUID
+        :raises Exception: if task deletion fails
+        """
+        pass
diff --git a/backend/app/models/Task.py b/backend/app/models/Task.py
new file mode 100644
index 00000000..9de664f4
--- /dev/null
+++ b/backend/app/models/Task.py
@@ -0,0 +1,75 @@
+import uuid
+from datetime import datetime
+from enum import Enum as PyEnum
+
+from sqlalchemy import Column, DateTime, ForeignKey
+from sqlalchemy import Enum as SQLEnum
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.orm import relationship
+
+from .Base import Base
+
+
+class TaskType(str, PyEnum):
+    INTAKE_FORM_REVIEW = "intake_form_review"
+    VOLUNTEER_APP_REVIEW = "volunteer_app_review"
+    PROFILE_UPDATE = "profile_update"
+    MATCHING = "matching"
+
+
+class TaskPriority(str, PyEnum):
+    NO_STATUS = "no_status"
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+
+
+class TaskStatus(str, PyEnum):
+    PENDING = "pending"
+    IN_PROGRESS = "in_progress"
+    COMPLETED = "completed"
+
+
+class Task(Base):
+    __tablename__ = "tasks"
+
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    participant_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
+    type = Column(
+        SQLEnum(
+            TaskType,
+            name="task_type_enum",
+            create_type=False,
+            values_callable=lambda enum_cls: [member.value for member in enum_cls],
+        ),
+        nullable=False,
+    )
+    priority = Column(
+        SQLEnum(
+            TaskPriority,
+            name="task_priority_enum",
+            create_type=False,
+            values_callable=lambda enum_cls: [member.value for member in enum_cls],
+        ),
+        nullable=False,
+        default=TaskPriority.NO_STATUS,
+    )
+    status = Column(
+        SQLEnum(
+            TaskStatus,
+            name="task_status_enum",
+            create_type=False,
+            values_callable=lambda enum_cls: [member.value for member in enum_cls],
+        ),
+        nullable=False,
+        default=TaskStatus.PENDING,
+    )
+    assignee_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
+    start_date = Column(DateTime, nullable=False, default=datetime.utcnow)
+    end_date = Column(DateTime, nullable=True)
+    created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
+    updated_at = Column(DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow)
+
+    # Relationships
+    participant = relationship("User", foreign_keys=[participant_id], backref="participant_tasks")
+    assignee = relationship("User", foreign_keys=[assignee_id], backref="assigned_tasks")
diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py
index 9d62a259..bc196d7d 100644
--- a/backend/app/models/__init__.py
+++ b/backend/app/models/__init__.py
@@ -19,6 +19,7 @@
 from .RankingPreference import RankingPreference
 from .Role import Role
 from .SuggestedTime import suggested_times
+from .Task import Task, TaskPriority, TaskStatus, TaskType
 from .TimeBlock import TimeBlock
 from .Treatment import Treatment
 from .User import FormStatus, User
@@ -43,6 +44,10 @@
     "Form",
     "FormSubmission",
     "FormStatus",
+    "Task",
+    "TaskType",
+    "TaskPriority",
+    "TaskStatus",
 ]
 
 log = logging.getLogger(LOGGER_NAME("models"))
diff --git a/backend/app/routes/task.py b/backend/app/routes/task.py
new file mode 100644
index 00000000..cded496a
--- /dev/null
+++ b/backend/app/routes/task.py
@@ -0,0 +1,163 @@
+from typing import Optional
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+
+from app.middleware.auth import has_roles
+from app.schemas.task import (
+    TaskAssignRequest,
+    TaskCreateRequest,
+    TaskListResponse,
+    TaskResponse,
+    TaskUpdateRequest,
+)
+from app.schemas.user import UserRole
+from app.services.implementations.task_service import TaskService
+from app.utilities.service_utils import get_task_service
+
+router = APIRouter(
+    prefix="/tasks",
+    tags=["tasks"],
+)
+
+
+@router.post("/", response_model=TaskResponse)
+async def create_task(
+    task: TaskCreateRequest,
+    task_service: TaskService = Depends(get_task_service),
+    authorized: bool = has_roles([UserRole.ADMIN]),
+):
+    """
+    Create a new task (admin only).
+    This endpoint can also be called internally from other services to automatically create tasks.
+    """
+    try:
+        return await task_service.create_task(task)
+    except HTTPException as http_ex:
+        raise http_ex
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.get("/", response_model=TaskListResponse)
+async def get_tasks(
+    status: Optional[str] = Query(None, description="Filter by task status"),
+    priority: Optional[str] = Query(None, description="Filter by task priority"),
+    task_type: Optional[str] = Query(None, description="Filter by task type"),
+    assignee_id: Optional[str] = Query(None, description="Filter by assignee ID"),
+    task_service: TaskService = Depends(get_task_service),
+    authorized: bool = has_roles([UserRole.ADMIN]),
+):
+    """
+    Get all tasks with optional filters (admin only)
+    """
+    try:
+        assignee_uuid = UUID(assignee_id) if assignee_id else None
+        tasks = await task_service.get_all_tasks(
+            status=status, priority=priority, task_type=task_type, assignee_id=assignee_uuid
+        )
+        return TaskListResponse(tasks=tasks, total=len(tasks))
+    except ValueError:
+        raise HTTPException(status_code=400, detail="Invalid assignee_id format")
+    except HTTPException as http_ex:
+        raise http_ex
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.get("/{task_id}", response_model=TaskResponse)
+async def get_task(
+    task_id: str,
+    task_service: TaskService = Depends(get_task_service),
+    authorized: bool = has_roles([UserRole.ADMIN]),
+):
+    """
+    Get a single task by ID (admin only)
+    """
+    try:
+        return await task_service.get_task_by_id(UUID(task_id))
+    except ValueError:
+        raise HTTPException(status_code=400, detail="Invalid task ID format")
+    except HTTPException as http_ex:
+        raise http_ex
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.put("/{task_id}", response_model=TaskResponse)
+async def update_task(
+    task_id: str,
+    task_update: TaskUpdateRequest,
+    task_service: TaskService = Depends(get_task_service),
+    authorized: bool = has_roles([UserRole.ADMIN]),
+):
+    """
+    Update a task (admin only)
+    """
+    try:
+        return await task_service.update_task(UUID(task_id), task_update)
+    except ValueError:
+        raise HTTPException(status_code=400, detail="Invalid task ID format")
+    except HTTPException as http_ex:
+        raise http_ex
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.put("/{task_id}/assign", response_model=TaskResponse)
+async def assign_task(
+    task_id: str,
+    assign_request: TaskAssignRequest,
+    task_service: TaskService = Depends(get_task_service),
+    authorized: bool = has_roles([UserRole.ADMIN]),
+):
+    """
+    Assign a task to an admin user (admin only)
+    """
+    try:
+        return await task_service.assign_task(UUID(task_id), assign_request.assignee_id)
+    except ValueError:
+        raise HTTPException(status_code=400, detail="Invalid task ID format")
+    except HTTPException as http_ex:
+        raise http_ex
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.put("/{task_id}/complete", response_model=TaskResponse)
+async def complete_task(
+    task_id: str,
+    task_service: TaskService = Depends(get_task_service),
+    authorized: bool = has_roles([UserRole.ADMIN]),
+):
+    """
+    Mark a task as completed (admin only)
+    """
+    try:
+        return await task_service.complete_task(UUID(task_id))
+    except ValueError:
+        raise HTTPException(status_code=400, detail="Invalid task ID format")
+    except HTTPException as http_ex:
+        raise http_ex
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.delete("/{task_id}")
+async def delete_task(
+    task_id: str,
+    task_service: TaskService = Depends(get_task_service),
+    authorized: bool = has_roles([UserRole.ADMIN]),
+):
+    """
+    Delete a task (admin only)
+    """
+    try:
+        await task_service.delete_task(UUID(task_id))
+        return {"message": "Task deleted successfully"}
+    except ValueError:
+        raise HTTPException(status_code=400, detail="Invalid task ID format")
+    except HTTPException as http_ex:
+        raise http_ex
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
diff --git a/backend/app/schemas/task.py b/backend/app/schemas/task.py
new file mode 100644
index 00000000..18eb13bb
--- /dev/null
+++ b/backend/app/schemas/task.py
@@ -0,0 +1,122 @@
+"""
+Pydantic schemas for task-related data validation and serialization.
+Handles task CRUD and response models for the API.
+"""
+
+from datetime import datetime
+from enum import Enum
+from typing import List, Optional
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class TaskType(str, Enum):
+    """
+    Enum for task types.
+    """
+
+    INTAKE_FORM_REVIEW = "intake_form_review"
+    VOLUNTEER_APP_REVIEW = "volunteer_app_review"
+    PROFILE_UPDATE = "profile_update"
+    MATCHING = "matching"
+
+
+class TaskPriority(str, Enum):
+    """
+    Enum for task priorities.
+    """
+
+    NO_STATUS = "no_status"
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+
+
+class TaskStatus(str, Enum):
+    """
+    Enum for task status.
+    """
+
+    PENDING = "pending"
+    IN_PROGRESS = "in_progress"
+    COMPLETED = "completed"
+
+
+class TaskBase(BaseModel):
+    """
+    Base schema for task model with common attributes shared across schemas.
+    """
+
+    participant_id: Optional[UUID] = None
+    type: TaskType
+    priority: TaskPriority = TaskPriority.NO_STATUS
+    status: TaskStatus = TaskStatus.PENDING
+    assignee_id: Optional[UUID] = None
+    start_date: datetime = Field(default_factory=datetime.utcnow)
+    end_date: Optional[datetime] = None
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class TaskCreateRequest(BaseModel):
+    """
+    Request schema for task creation.
+    """
+
+    participant_id: Optional[UUID] = None
+    type: TaskType
+    priority: TaskPriority = TaskPriority.NO_STATUS
+    assignee_id: Optional[UUID] = None
+    start_date: Optional[datetime] = None
+    end_date: Optional[datetime] = None
+
+
+class TaskUpdateRequest(BaseModel):
+    """
+    Request schema for task updates, all fields optional.
+    """
+
+    participant_id: Optional[UUID] = None
+    type: Optional[TaskType] = None
+    priority: Optional[TaskPriority] = None
+    status: Optional[TaskStatus] = None
+    assignee_id: Optional[UUID] = None
+    start_date: Optional[datetime] = None
+    end_date: Optional[datetime] = None
+
+
+class TaskAssignRequest(BaseModel):
+    """
+    Request schema for assigning a task to an admin.
+    """
+
+    assignee_id: UUID
+
+
+class TaskResponse(BaseModel):
+    """
+    Response schema for task data.
+    """
+
+    id: UUID
+    participant_id: Optional[UUID]
+    type: TaskType
+    priority: TaskPriority
+    status: TaskStatus
+    assignee_id: Optional[UUID]
+    start_date: datetime
+    end_date: Optional[datetime]
+    created_at: datetime
+    updated_at: datetime
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class TaskListResponse(BaseModel):
+    """
+    Response schema for listing tasks.
+    """
+
+    tasks: List[TaskResponse]
+    total: int
diff --git a/backend/app/server.py b/backend/app/server.py
index 41def270..ed62fd45 100644
--- a/backend/app/server.py
+++ b/backend/app/server.py
@@ -8,7 +8,7 @@
 
 from . import models
 from .middleware.auth_middleware import AuthMiddleware
-from .routes import auth, availability, intake, match, matching, ranking, send_email, suggested_times, test, user
+from .routes import auth, availability, intake, match, matching, ranking, send_email, suggested_times, task, test, user
 from .utilities.constants import LOGGER_NAME
 from .utilities.firebase_init import initialize_firebase
 from .utilities.ses.ses_init import ensure_ses_templates
@@ -72,6 +72,7 @@ async def lifespan(_: FastAPI):
 app.include_router(intake.router)
 app.include_router(ranking.router)
 app.include_router(send_email.router)
+app.include_router(task.router)
 app.include_router(test.router)
 
 
diff --git a/backend/app/services/implementations/task_service.py b/backend/app/services/implementations/task_service.py
new file mode 100644
index 00000000..82fb20f6
--- /dev/null
+++ b/backend/app/services/implementations/task_service.py
@@ -0,0 +1,245 @@
+import logging
+from datetime import datetime
+from typing import List, Optional
+from uuid import UUID
+
+from fastapi import HTTPException
+from sqlalchemy.orm import Session
+
+from app.interfaces.task_service import ITaskService
+from app.models import Task, TaskPriority, TaskStatus, TaskType, User
+from app.schemas.task import TaskCreateRequest, TaskResponse, TaskUpdateRequest
+from app.utilities.constants import LOGGER_NAME
+
+
+class TaskService(ITaskService):
+    def __init__(self, db: Session):
+        self.db = db
+        self.logger = logging.getLogger(LOGGER_NAME("task_service"))
+
+    async def create_task(self, task: TaskCreateRequest) -> TaskResponse:
+        """
+        Create a new task in the database
+        """
+        try:
+            # Validate participant exists if provided
+            if task.participant_id:
+                participant = self.db.query(User).filter(User.id == task.participant_id).first()
+                if not participant:
+                    raise HTTPException(status_code=404, detail="Participant not found")
+
+            # Validate assignee is an admin if provided
+            if task.assignee_id:
+                assignee = self.db.query(User).filter(User.id == task.assignee_id, User.role_id == 3).first()
+                if not assignee:
+                    raise HTTPException(status_code=404, detail="Assignee must be an admin user")
+
+            db_task = Task(
+                participant_id=task.participant_id,
+                type=TaskType(task.type),
+                priority=TaskPriority(task.priority),
+                assignee_id=task.assignee_id,
+                start_date=task.start_date or datetime.utcnow(),
+                end_date=task.end_date,
+            )
+
+            self.db.add(db_task)
+            self.db.commit()
+            self.db.refresh(db_task)
+
+            self.logger.info(f"Created task {db_task.id} of type {task.type}")
+            return TaskResponse.model_validate(db_task)
+
+        except HTTPException:
+            raise
+        except Exception as e:
+            self.db.rollback()
+            self.logger.error(f"Error creating task: {str(e)}")
+            raise HTTPException(status_code=500, detail=str(e))
+
+    async def get_task_by_id(self, task_id: UUID) -> TaskResponse:
+        """
+        Get a task by its ID
+        """
+        try:
+            task = self.db.query(Task).filter(Task.id == task_id).first()
+            if not task:
+                raise HTTPException(status_code=404, detail="Task not found")
+            return TaskResponse.model_validate(task)
+        except HTTPException:
+            raise
+        except Exception as e:
+            self.logger.error(f"Error retrieving task {task_id}: {str(e)}")
+            raise HTTPException(status_code=500, detail=str(e))
+
+    async def get_all_tasks(
+        self,
+        status: Optional[str] = None,
+        priority: Optional[str] = None,
+        task_type: Optional[str] = None,
+        assignee_id: Optional[UUID] = None,
+    ) -> List[TaskResponse]:
+        """
+        Get all tasks with optional filters
+        """
+        try:
+            query = self.db.query(Task)
+
+            # Apply filters if provided
+            if status:
+                try:
+                    query = query.filter(Task.status == TaskStatus(status))
+                except ValueError:
+                    raise HTTPException(status_code=400, detail="Invalid status value")
+
+            if priority:
+                try:
+                    query = query.filter(Task.priority == TaskPriority(priority))
+                except ValueError:
+                    raise HTTPException(status_code=400, detail="Invalid priority value")
+
+            if task_type:
+                try:
+                    query = query.filter(Task.type == TaskType(task_type))
+                except ValueError:
+                    raise HTTPException(status_code=400, detail="Invalid task type value")
+
+            if assignee_id:
+                query = query.filter(Task.assignee_id == assignee_id)
+
+            tasks = query.order_by(Task.created_at.desc()).all()
+            return [TaskResponse.model_validate(task) for task in tasks]
+        except HTTPException:
+            raise
+        except Exception as e:
+            self.logger.error(f"Error retrieving tasks: {str(e)}")
+            raise HTTPException(status_code=500, detail=str(e))
+
+    async def update_task(self, task_id: UUID, task_update: TaskUpdateRequest) -> TaskResponse:
+        """
+        Update a task
+        """
+        try:
+            db_task = self.db.query(Task).filter(Task.id == task_id).first()
+            if not db_task:
+                raise HTTPException(status_code=404, detail="Task not found")
+
+            # Update provided fields only
+            update_data = task_update.model_dump(exclude_unset=True)
+
+            # Validate participant if being updated
+            if "participant_id" in update_data and update_data["participant_id"]:
+                participant = self.db.query(User).filter(User.id == update_data["participant_id"]).first()
+                if not participant:
+                    raise HTTPException(status_code=404, detail="Participant not found")
+
+            # Validate assignee is admin if being updated
+            if "assignee_id" in update_data and update_data["assignee_id"]:
+                assignee = self.db.query(User).filter(User.id == update_data["assignee_id"], User.role_id == 3).first()
+                if not assignee:
+                    raise HTTPException(status_code=404, detail="Assignee must be an admin user")
+
+            # Convert enum strings to enum types if needed
+            if "type" in update_data:
+                update_data["type"] = TaskType(update_data["type"])
+            if "priority" in update_data:
+                update_data["priority"] = TaskPriority(update_data["priority"])
+            if "status" in update_data:
+                update_data["status"] = TaskStatus(update_data["status"])
+
+            for field, value in update_data.items():
+                setattr(db_task, field, value)
+
+            db_task.updated_at = datetime.utcnow()
+
+            self.db.commit()
+            self.db.refresh(db_task)
+
+            self.logger.info(f"Updated task {task_id}")
+            return TaskResponse.model_validate(db_task)
+
+        except HTTPException:
+            raise
+        except ValueError as e:
+            raise HTTPException(status_code=400, detail=f"Invalid enum value: {str(e)}")
+        except Exception as e:
+            self.db.rollback()
+            self.logger.error(f"Error updating task {task_id}: {str(e)}")
+            raise HTTPException(status_code=500, detail=str(e))
+
+    async def assign_task(self, task_id: UUID, assignee_id: UUID) -> TaskResponse:
+        """
+        Assign a task to an admin user
+        """
+        try:
+            db_task = self.db.query(Task).filter(Task.id == task_id).first()
+            if not db_task:
+                raise HTTPException(status_code=404, detail="Task not found")
+
+            # Validate assignee is an admin
+            assignee = self.db.query(User).filter(User.id == assignee_id, User.role_id == 3).first()
+            if not assignee:
+                raise HTTPException(status_code=404, detail="Assignee must be an admin user")
+
+            db_task.assignee_id = assignee_id
+            db_task.updated_at = datetime.utcnow()
+
+            self.db.commit()
+            self.db.refresh(db_task)
+
+            self.logger.info(f"Assigned task {task_id} to admin {assignee_id}")
+            return TaskResponse.model_validate(db_task)
+
+        except HTTPException:
+            raise
+        except Exception as e:
+            self.db.rollback()
+            self.logger.error(f"Error assigning task {task_id}: {str(e)}")
+            raise HTTPException(status_code=500, detail=str(e))
+
+    async def complete_task(self, task_id: UUID) -> TaskResponse:
+        """
+        Mark a task as completed
+        """
+        try:
+            db_task = self.db.query(Task).filter(Task.id == task_id).first()
+            if not db_task:
+                raise HTTPException(status_code=404, detail="Task not found")
+
+            db_task.status = TaskStatus.COMPLETED
+            db_task.end_date = datetime.utcnow()
+            db_task.updated_at = datetime.utcnow()
+
+            self.db.commit()
+            self.db.refresh(db_task)
+
+            self.logger.info(f"Completed task {task_id}")
+            return TaskResponse.model_validate(db_task)
+
+        except HTTPException:
+            raise
+        except Exception as e:
+            self.db.rollback()
+            self.logger.error(f"Error completing task {task_id}: {str(e)}")
+            raise HTTPException(status_code=500, detail=str(e))
+
+    async def delete_task(self, task_id: UUID) -> None:
+        """
+        Delete a task
+        """
+        try:
+            db_task = self.db.query(Task).filter(Task.id == task_id).first()
+            if not db_task:
+                raise HTTPException(status_code=404, detail="Task not found")
+
+            self.db.delete(db_task)
+            self.db.commit()
+
+            self.logger.info(f"Deleted task {task_id}")
+
+        except HTTPException:
+            raise
+        except Exception as e:
+            self.db.rollback()
+            self.logger.error(f"Error deleting task {task_id}: {str(e)}")
+            raise HTTPException(status_code=500, detail=str(e))
diff --git a/backend/app/utilities/service_utils.py b/backend/app/utilities/service_utils.py
index a362a99c..23c8dc08 100644
--- a/backend/app/utilities/service_utils.py
+++ b/backend/app/utilities/service_utils.py
@@ -4,6 +4,7 @@
 from sqlalchemy.orm import Session
 
 from ..services.implementations.auth_service import AuthService
+from ..services.implementations.task_service import TaskService
 from ..services.implementations.user_service import UserService
 from .db_utils import get_db
 
@@ -15,3 +16,7 @@ def get_user_service(db: Session = Depends(get_db)):
 def get_auth_service(user_service: UserService = Depends(get_user_service)):
     logger = logging.getLogger(__name__)
     return AuthService(logger=logger, user_service=user_service)
+
+
+def get_task_service(db: Session = Depends(get_db)):
+    return TaskService(db)
diff --git a/backend/migrations/versions/0fb019b7af03_add_tasks_table_with_type_priority_.py b/backend/migrations/versions/0fb019b7af03_add_tasks_table_with_type_priority_.py
new file mode 100644
index 00000000..d44f270d
--- /dev/null
+++ b/backend/migrations/versions/0fb019b7af03_add_tasks_table_with_type_priority_.py
@@ -0,0 +1,55 @@
+"""Add tasks table with type, priority, status enums
+
+Revision ID: 0fb019b7af03
+Revises: b56e0bf600a2
+Create Date: 2025-10-02 21:01:02.761026
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "0fb019b7af03"
+down_revision: Union[str, None] = "b56e0bf600a2"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "tasks",
+        sa.Column("id", sa.UUID(), nullable=False),
+        sa.Column("participant_id", sa.UUID(), nullable=True),
+        sa.Column(
+            "type",
+            sa.Enum("intake_form_review", "volunteer_app_review", "profile_update", "matching", name="task_type_enum"),
+            nullable=False,
+        ),
+        sa.Column("priority", sa.Enum("no_status", "low", "medium", "high", name="task_priority_enum"), nullable=False),
+        sa.Column("status", sa.Enum("pending", "in_progress", "completed", name="task_status_enum"), nullable=False),
+        sa.Column("assignee_id", sa.UUID(), nullable=True),
+        sa.Column("start_date", sa.DateTime(), nullable=False),
+        sa.Column("end_date", sa.DateTime(), nullable=True),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["assignee_id"],
+            ["users.id"],
+        ),
+        sa.ForeignKeyConstraint(
+            ["participant_id"],
+            ["users.id"],
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("tasks")
+    # ### end Alembic commands ###