diff --git a/AGENTS.md b/AGENTS.md
index 89be23d62..6643a7312 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -78,6 +78,28 @@ async def process_data(data: dict) -> Result:
         raise ValidationError(f"Missing required field: {e}") from e
 ```
 
+**SQL Safety patterns**:
+- NEVER use string formatting for SQL queries (f-strings, .format(), string concatenation)
+- ALWAYS use parameterized queries with `$1, $2, $3` placeholders
+- Use `SafeQueryBuilder` from `agents_api.queries.sql_utils` for complex dynamic queries
+- Use `safe_format_query` for simple queries with validated ORDER BY clauses
+- Validate all identifiers with `sanitize_identifier()` when needed
+- Whitelist allowed sort fields and directions
+
+Example:
+```python
+from agents_api.queries.sql_utils import SafeQueryBuilder
+
+# Good - using SafeQueryBuilder
+builder = SafeQueryBuilder("SELECT * FROM agents WHERE developer_id = $1", [dev_id])
+builder.add_condition(" AND status = {}", status)
+builder.add_order_by(sort_field, direction, allowed_fields={"created_at", "name"})
+query, params = builder.build()
+
+# Bad - NEVER do this
+query = f"SELECT * FROM agents WHERE status = '{status}' ORDER BY {sort_field}"
+```
+
 ---
 
 ## 4. Project layout & Core Components
@@ -213,6 +235,7 @@ async def create_entry(
 *   Large AI refactors in a single commit (makes `git bisect` difficult).
 *   Delegating test/spec writing entirely to AI (can lead to false confidence).
 *   **Note about `src/`**: Only the `cli` component has a `src/` directory. For `agents-api`, code is directly in `agents_api/`. Follow the existing pattern for each component.
+*   **SQL Injection vulnerabilities**: Using string formatting (f-strings, .format(), %) for SQL queries instead of parameterized queries and the SQL safety utilities in `agents_api/queries/sql_utils.py`.
 
 ---
 
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index d697fba4a..28323fc39 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -179,6 +179,27 @@ To get a comprehensive understanding of Julep, we recommend exploring the codeba
    - Add new tests for new functionality
    - Ensure all tests pass before submitting your changes
 
+### Security Guidelines
+
+When contributing code that interacts with databases:
+
+1. **SQL Injection Prevention**
+   - NEVER use string formatting for SQL queries (f-strings, .format(), %)
+   - ALWAYS use the SQL safety utilities from `agents-api/agents_api/queries/sql_utils.py`
+   - Use `SafeQueryBuilder` for complex queries with dynamic conditions
+   - Use `safe_format_query` for simple queries with ORDER BY clauses
+   - All user inputs must be parameterized using placeholder syntax ($1, $2, etc.)
+
+2. **Input Validation**
+   - Validate and sanitize all user inputs before using in queries
+   - Use whitelisting for sort fields and directions
+   - Check identifier patterns with `sanitize_identifier()`
+
+3. **Testing Security**
+   - Add tests for any new query functions that handle user input
+   - See `agents-api/tests/test_sql_injection_prevention.py` for examples
+   - Ensure your code passes SQL injection prevention tests
+
 5. **Submit a Pull Request**
    - Provide a clear description of your changes
    - Reference any related issues
diff --git a/agents-api/AGENTS.md b/agents-api/AGENTS.md
index 7a769fdd6..894138a0f 100644
--- a/agents-api/AGENTS.md
+++ b/agents-api/AGENTS.md
@@ -75,3 +75,35 @@ Key Uses
 - Expression validation checks syntax, undefined names, unsafe operations
 - Task validation checks all expressions in workflow steps
 - Security: Sandbox with limited function/module access
+
+## SQL Safety Requirements
+- ALWAYS use the SQL utilities from `agents_api/queries/sql_utils.py`
+- NEVER use string formatting (f-strings, .format(), %) for SQL queries
+- Use `SafeQueryBuilder` for complex queries with dynamic WHERE, ORDER BY, LIMIT
+- Use `safe_format_query` for simple ORDER BY validation
+- All sort fields must be whitelisted explicitly
+- Run SQL injection tests: `poe test --search "sql_injection_prevention"`
+
+### Example Usage
+```python
+# For complex queries
+from agents_api.queries.sql_utils import SafeQueryBuilder
+
+builder = SafeQueryBuilder("SELECT * FROM docs WHERE developer_id = $1", [dev_id])
+if metadata_filter:
+    for key, value in metadata_filter.items():
+        builder.add_condition(" AND metadata->>{}::text = {}", key, str(value))
+builder.add_order_by("created_at", "desc", allowed_fields={"created_at", "updated_at"})
+builder.add_limit_offset(limit, offset)
+query, params = builder.build()
+
+# For simple ORDER BY queries
+from agents_api.queries.sql_utils import safe_format_query
+
+query = safe_format_query(
+    "SELECT * FROM entries ORDER BY {sort_by} {direction}",
+    sort_by="created_at",
+    direction="desc",
+    allowed_sort_fields={"created_at", "timestamp"}
+)
+```
diff --git a/agents-api/README.md b/agents-api/README.md
index 66418bc52..5b98bb2fb 100644
--- a/agents-api/README.md
+++ b/agents-api/README.md
@@ -10,6 +10,10 @@ The `agents-api` project serves as the foundation of the agent management system
 
 The `models` module encapsulates all data interactions with the CozoDB database, providing a structured way to perform CRUD operations and other specific data manipulations across various entities.
 
+### Queries
+
+The `queries` module contains database query builders organized by resource types. It includes critical SQL safety utilities in `sql_utils.py` to prevent SQL injection attacks through parameterized queries and input validation.
+
 ### Routers
 
 The `routers` module handles HTTP routing for different parts of the application, directing incoming HTTP requests to the appropriate handler functions.
diff --git a/agents-api/agents_api/queries/AGENTS.md b/agents-api/agents_api/queries/AGENTS.md
index 9c5fe4255..538627ec8 100644
--- a/agents-api/agents_api/queries/AGENTS.md
+++ b/agents-api/agents_api/queries/AGENTS.md
@@ -9,6 +9,7 @@ Key Points
 - Add new queries in `queries/` and index them if needed.
 - Tests reside under `agents-api/tests/`.
 - Add `AIDEV-NOTE` anchors at the top of query modules to clarify module purpose.
+- **ALWAYS use SQL safety utilities** from `sql_utils.py` to prevent SQL injection.
 
 # Queries
 
@@ -70,3 +71,53 @@ Key Points
 - `prepare_execution_input`: Builds inputs for Temporal workflows
 - `create_execution_transition`: Records state changes in executions
 - `search_docs_hybrid`: Combined embedding and text search
+
+## SQL Safety and Security
+
+### SQL Injection Prevention
+All queries MUST use the utilities from `sql_utils.py` to prevent SQL injection attacks:
+
+1. **SafeQueryBuilder**: For complex queries with dynamic conditions
+   ```python
+   from ..sql_utils import SafeQueryBuilder
+   
+   builder = SafeQueryBuilder(base_query, initial_params)
+   builder.add_condition(" AND status = {}", status_value)
+   builder.add_order_by("created_at", "desc", allowed_fields={"created_at", "updated_at"})
+   builder.add_limit_offset(limit, offset)
+   query, params = builder.build()
+   ```
+
+2. **safe_format_query**: For simple queries with ORDER BY clauses
+   ```python
+   from ..sql_utils import safe_format_query
+   
+   query = safe_format_query(
+       query_template,
+       sort_by="created_at",
+       direction="desc",
+       allowed_sort_fields={"created_at", "updated_at", "name"},
+       table_prefix="t."
+   )
+   ```
+
+3. **validate_sort_field** and **validate_sort_direction**: For validating user inputs
+   - Uses whitelisting approach
+   - Validates against SQL identifier patterns
+   - Prevents SQL keyword injection
+
+### Never Do This
+- ❌ NEVER use f-strings for SQL: `f"SELECT * FROM {table} WHERE {field} = '{value}'"`
+- ❌ NEVER use .format() for SQL: `"SELECT * FROM {} WHERE {}".format(table, condition)`
+- ❌ NEVER concatenate user input into SQL: `query + " ORDER BY " + user_input`
+
+### Always Do This
+- ✅ Use parameterized queries: `$1, $2, $3` placeholders
+- ✅ Use SafeQueryBuilder for complex dynamic queries
+- ✅ Validate all identifiers with sanitize_identifier()
+- ✅ Whitelist allowed sort fields and directions
+
+### Testing
+- SQL injection prevention tests are in `tests/test_sql_injection_prevention.py`
+- Run with: `poe test --search "sql_injection_prevention"`
+- 27 comprehensive tests covering various attack vectors
diff --git a/agents-api/agents_api/queries/agents/list_agents.py b/agents-api/agents_api/queries/agents/list_agents.py
index 122b64454..82f35afd8 100644
--- a/agents-api/agents_api/queries/agents/list_agents.py
+++ b/agents-api/agents_api/queries/agents/list_agents.py
@@ -89,10 +89,9 @@ async def list_agents(
         direction,
     ]
 
-    # Handle metadata filter differently - using JSONB containment
-    agent_query = raw_query.format(
-        metadata_filter_query="AND a.metadata @> $6::jsonb" if metadata_filter else "",
-    )
+    # AIDEV-NOTE: Build metadata filter query safely to prevent SQL injection
+    metadata_filter_query = "AND a.metadata @> $6::jsonb" if metadata_filter else ""
+    agent_query = raw_query.replace("{metadata_filter_query}", metadata_filter_query)
 
     # If we have metadata filters, safely add them as a parameter
     if metadata_filter:
diff --git a/agents-api/agents_api/queries/docs/bulk_delete_docs.py b/agents-api/agents_api/queries/docs/bulk_delete_docs.py
index df5fe1978..61a51a55d 100644
--- a/agents-api/agents_api/queries/docs/bulk_delete_docs.py
+++ b/agents-api/agents_api/queries/docs/bulk_delete_docs.py
@@ -52,7 +52,9 @@ async def bulk_delete_docs(
         params, metadata_filter if not delete_all else {}, table_alias="d."
     )
 
-    query = f"""
+    # AIDEV-NOTE: Build query with proper parameter placeholders to avoid SQL injection
+    query = (
+        """
     WITH deleted_docs AS (
         DELETE FROM docs d
         WHERE d.developer_id = $1
@@ -63,7 +65,9 @@ async def bulk_delete_docs(
                 AND o.developer_id = d.developer_id
                 AND o.owner_type = $2
                 AND o.owner_id = $3
-                {metadata_conditions}
+                """
+        + metadata_conditions
+        + """
             )
         RETURNING d.doc_id
     ),
@@ -76,5 +80,6 @@ async def bulk_delete_docs(
     )
     SELECT doc_id FROM deleted_docs;
     """
+    )
 
     return query, params
diff --git a/agents-api/agents_api/queries/docs/list_docs.py b/agents-api/agents_api/queries/docs/list_docs.py
index 39aaba00e..85f024699 100644
--- a/agents-api/agents_api/queries/docs/list_docs.py
+++ b/agents-api/agents_api/queries/docs/list_docs.py
@@ -11,8 +11,8 @@
 
 from ...autogen.openapi_model import Doc
 from ...common.utils.db_exceptions import common_db_exceptions
+from ..sql_utils import SafeQueryBuilder
 from ..utils import (
-    build_metadata_filter_conditions,
     make_num_validator,
     pg_query,
     rewrap_exceptions,
@@ -105,18 +105,19 @@ async def list_docs(
 
     # AIDEV-NOTE: avoid mutable default; initialize metadata_filter
     metadata_filter = metadata_filter if metadata_filter is not None else {}
-    # Start with the base query
-    query = base_docs_query
-    params = [developer_id, include_without_embeddings, owner_type, owner_id]
 
-    # Add metadata filtering before GROUP BY using the utility function with table alias
-    metadata_conditions, params = build_metadata_filter_conditions(
-        params, metadata_filter, table_alias="d."
+    # Build query using SafeQueryBuilder to prevent SQL injection
+    builder = SafeQueryBuilder(
+        base_docs_query, [developer_id, include_without_embeddings, owner_type, owner_id]
     )
-    query += metadata_conditions
+
+    # Add metadata filtering using SafeQueryBuilder's condition system
+    if metadata_filter:
+        for key, value in metadata_filter.items():
+            builder.add_condition(" AND d.metadata->>{}::text = {}", key, str(value))
 
     # Add GROUP BY clause
-    query += """
+    builder.add_raw_condition("""
     GROUP BY
         d.doc_id,
         d.developer_id,
@@ -126,12 +127,12 @@ async def list_docs(
         d.embedding_dimensions,
         d.language,
         d.metadata,
-        d.created_at"""
+        d.created_at""")
 
-    # Add sorting and pagination
-    query += (
-        f" ORDER BY {sort_by} {direction} LIMIT ${len(params) + 1} OFFSET ${len(params) + 2}"
+    # Add sorting and pagination with validation
+    builder.add_order_by(
+        sort_by, direction, allowed_fields={"created_at", "updated_at"}, table_prefix=""
     )
-    params.extend([limit, offset])
+    builder.add_limit_offset(limit, offset)
 
-    return query, params
+    return builder.build()
diff --git a/agents-api/agents_api/queries/entries/list_entries.py b/agents-api/agents_api/queries/entries/list_entries.py
index 3d655c263..66e6f3ec3 100644
--- a/agents-api/agents_api/queries/entries/list_entries.py
+++ b/agents-api/agents_api/queries/entries/list_entries.py
@@ -9,6 +9,7 @@
 from ...common.utils.datetime import utcnow
 from ...common.utils.db_exceptions import common_db_exceptions
 from ...metrics.counters import query_metrics
+from ..sql_utils import safe_format_query
 from ..utils import make_num_validator, pg_query, rewrap_exceptions, wrap_in_class
 
 # Query for checking if the session exists
@@ -43,7 +44,7 @@
     AND (er.relation IS NULL OR er.relation != ALL($6))
     AND e.created_at >= $7
     AND e.created_at >= (select created_at from sessions where session_id = $1 LIMIT 1)
-ORDER BY e.{sort_by} {direction} -- safe to interpolate
+ORDER BY e.{sort_by} {direction}
 LIMIT $3
 OFFSET $4;
 """
@@ -96,9 +97,14 @@ async def list_entries(
         allowed_sources if allowed_sources is not None else ["api_request", "api_response"]
     )
     exclude_relations = exclude_relations if exclude_relations is not None else []
-    query = list_entries_query.format(
+
+    # AIDEV-NOTE: Use safe_format_query to prevent SQL injection in ORDER BY clause
+    query = safe_format_query(
+        list_entries_query,
         sort_by=sort_by,
         direction=direction,
+        allowed_sort_fields={"created_at", "timestamp"},
+        table_prefix="",
     )
 
     # Parameters for the entry query
diff --git a/agents-api/agents_api/queries/files/list_files.py b/agents-api/agents_api/queries/files/list_files.py
index 7983d5964..7bbbf2eca 100644
--- a/agents-api/agents_api/queries/files/list_files.py
+++ b/agents-api/agents_api/queries/files/list_files.py
@@ -11,6 +11,7 @@
 
 from ...autogen.openapi_model import File
 from ...common.utils.db_exceptions import common_db_exceptions
+from ..sql_utils import SafeQueryBuilder
 from ..utils import make_num_validator, pg_query, rewrap_exceptions, wrap_in_class
 
 # Base query for listing files
@@ -62,25 +63,22 @@ async def list_files(
     """
     Lists files with optional owner and project filtering, pagination, and sorting.
     """
-    # Start with the base query
-    query = base_files_query
-    params = [developer_id]
-    param_index = 2
+    # Build query using SafeQueryBuilder to prevent SQL injection
+    builder = SafeQueryBuilder(base_files_query, [developer_id])
 
     # Add owner filtering
     if owner_type and owner_id:
-        query += f" AND fo.owner_type = ${param_index} AND fo.owner_id = ${param_index + 1}"
-        params.extend([owner_type, owner_id])
-        param_index += 2
+        builder.add_condition(" AND fo.owner_type = {}", owner_type)
+        builder.add_condition(" AND fo.owner_id = {}", owner_id)
 
     # Add project filtering
     if project:
-        query += f" AND p.canonical_name = ${param_index}"
-        params.append(project)
-        param_index += 1
+        builder.add_condition(" AND p.canonical_name = {}", project)
 
-    # Add sorting and pagination
-    query += f" ORDER BY f.{sort_by} {direction} LIMIT ${param_index} OFFSET ${param_index + 1}"
-    params.extend([limit, offset])
+    # Add sorting and pagination with validation
+    builder.add_order_by(
+        sort_by, direction, allowed_fields={"created_at", "updated_at"}, table_prefix="f."
+    )
+    builder.add_limit_offset(limit, offset)
 
-    return query, params
+    return builder.build()
diff --git a/agents-api/agents_api/queries/sql_utils.py b/agents-api/agents_api/queries/sql_utils.py
new file mode 100644
index 000000000..19bd4ee86
--- /dev/null
+++ b/agents-api/agents_api/queries/sql_utils.py
@@ -0,0 +1,298 @@
+"""
+SQL utilities for safe query construction and SQL injection prevention.
+
+This module provides utilities to safely construct SQL queries, preventing SQL injection
+attacks through proper validation and sanitization of identifiers and dynamic query parts.
+
+AIDEV-NOTE: Critical security module - prevents SQL injection in dynamic queries.
+Use SafeQueryBuilder for complex queries, safe_format_query for simple ORDER BY.
+Never use f-strings or .format() with user input in SQL queries!
+"""
+
+import re
+from typing import Any, Literal
+
+from beartype import beartype
+from fastapi import HTTPException
+
+# AIDEV-NOTE: Whitelist of allowed column names for sorting across different tables
+ALLOWED_SORT_COLUMNS = {
+    "created_at",
+    "updated_at",
+    "timestamp",
+    "name",
+    "title",
+}
+
+# AIDEV-NOTE: Regex pattern for valid SQL identifiers (alphanumeric + underscore, not starting with digit)
+SQL_IDENTIFIER_PATTERN = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$")
+
+
+@beartype
+def sanitize_identifier(identifier: str, identifier_type: str = "column") -> str:
+    """
+    Validates and sanitizes SQL identifiers (table names, column names) to prevent SQL injection.
+
+    Args:
+        identifier: The identifier to validate
+        identifier_type: Type of identifier for error messages (e.g., "column", "table")
+
+    Returns:
+        The validated identifier
+
+    Raises:
+        HTTPException: If the identifier is invalid
+    """
+    if not identifier:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Invalid {identifier_type} name: cannot be empty",
+        )
+
+    # Check against regex pattern
+    if not SQL_IDENTIFIER_PATTERN.match(identifier):
+        raise HTTPException(
+            status_code=400,
+            detail=f"Invalid {identifier_type} name: '{identifier}'. Must contain only letters, numbers, and underscores, and cannot start with a number.",
+        )
+
+    # Check length (PostgreSQL limit is 63 characters)
+    if len(identifier) > 63:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Invalid {identifier_type} name: '{identifier}' is too long (max 63 characters)",
+        )
+
+    # Check for SQL keywords (basic list, can be extended)
+    sql_keywords = {
+        "select",
+        "insert",
+        "update",
+        "delete",
+        "drop",
+        "create",
+        "alter",
+        "table",
+        "from",
+        "where",
+        "join",
+        "union",
+        "grant",
+        "revoke",
+    }
+    if identifier.lower() in sql_keywords:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Invalid {identifier_type} name: '{identifier}' is a reserved SQL keyword",
+        )
+
+    return identifier
+
+
+@beartype
+def validate_sort_field(
+    field: str, allowed_fields: set[str] | None = None, table_prefix: str = ""
+) -> str:
+    """
+    Validates sort field names against a whitelist to prevent SQL injection.
+
+    Args:
+        field: The field name to validate
+        allowed_fields: Set of allowed field names (defaults to ALLOWED_SORT_COLUMNS)
+        table_prefix: Optional table prefix to prepend (e.g., "e." for "e.created_at")
+
+    Returns:
+        The validated field name with optional table prefix
+
+    Raises:
+        HTTPException: If the field is not in the allowed list
+    """
+    allowed = allowed_fields or ALLOWED_SORT_COLUMNS
+
+    if field not in allowed:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Invalid sort field: '{field}'. Allowed fields: {', '.join(sorted(allowed))}",
+        )
+
+    # Additional validation using sanitize_identifier
+    sanitize_identifier(field, "sort field")
+
+    return f"{table_prefix}{field}" if table_prefix else field
+
+
+@beartype
+def validate_sort_direction(direction: str) -> Literal["ASC", "DESC"]:
+    """
+    Validates sort direction to prevent SQL injection.
+
+    Args:
+        direction: The sort direction to validate
+
+    Returns:
+        The validated direction in uppercase
+
+    Raises:
+        HTTPException: If the direction is invalid
+    """
+    direction_upper = direction.upper()
+    if direction_upper not in ("ASC", "DESC"):
+        raise HTTPException(
+            status_code=400,
+            detail=f"Invalid sort direction: '{direction}'. Must be 'asc' or 'desc'",
+        )
+    return direction_upper  # type: ignore
+
+
+class SafeQueryBuilder:
+    """
+    A utility class for safely building dynamic SQL queries.
+
+    This class helps construct SQL queries with proper parameterization,
+    avoiding SQL injection vulnerabilities.
+    """
+
+    def __init__(self, base_query: str, initial_params: list[Any] | None = None):
+        """
+        Initialize the query builder.
+
+        Args:
+            base_query: The base SQL query string
+            initial_params: Initial list of query parameters
+        """
+        self.query_parts: list[str] = [base_query]
+        self.params: list[Any] = initial_params or []
+        self._param_counter = len(self.params)
+
+    def add_condition(self, condition: str, *params: Any) -> "SafeQueryBuilder":
+        """
+        Add a WHERE condition with parameterized values.
+
+        Args:
+            condition: The condition string with placeholders (e.g., "user_id = {}" or "date BETWEEN {} AND {}")
+            *params: The parameter values
+
+        Returns:
+            Self for method chaining
+        """
+        # Count the number of {} placeholders in the condition
+        placeholder_count = condition.count("{}")
+        if placeholder_count != len(params):
+            msg = f"Expected {placeholder_count} parameters, got {len(params)}"
+            raise ValueError(msg)
+
+        # Replace each {} with the appropriate parameter placeholder
+        formatted_condition = condition
+        for param in params:
+            self._param_counter += 1
+            formatted_condition = formatted_condition.replace(
+                "{}", f"${self._param_counter}", 1
+            )
+            self.params.append(param)
+
+        self.query_parts.append(formatted_condition)
+        return self
+
+    def add_raw_condition(self, condition: str) -> "SafeQueryBuilder":
+        """
+        Add a raw SQL condition (use with caution, ensure it's validated).
+
+        Args:
+            condition: The raw SQL condition
+
+        Returns:
+            Self for method chaining
+        """
+        self.query_parts.append(condition)
+        return self
+
+    def add_order_by(
+        self,
+        field: str,
+        direction: str = "ASC",
+        allowed_fields: set[str] | None = None,
+        table_prefix: str = "",
+    ) -> "SafeQueryBuilder":
+        """
+        Add ORDER BY clause with validation.
+
+        Args:
+            field: The field to sort by
+            direction: Sort direction (ASC/DESC)
+            allowed_fields: Allowed field names for sorting
+            table_prefix: Optional table prefix
+
+        Returns:
+            Self for method chaining
+        """
+        safe_field = validate_sort_field(field, allowed_fields, table_prefix)
+        safe_direction = validate_sort_direction(direction)
+        self.query_parts.append(f" ORDER BY {safe_field} {safe_direction}")
+        return self
+
+    def add_limit_offset(self, limit: int, offset: int) -> "SafeQueryBuilder":
+        """
+        Add LIMIT and OFFSET clauses.
+
+        Args:
+            limit: Maximum number of rows
+            offset: Number of rows to skip
+
+        Returns:
+            Self for method chaining
+        """
+        self._param_counter += 1
+        limit_param = f"${self._param_counter}"
+        self._param_counter += 1
+        offset_param = f"${self._param_counter}"
+
+        self.query_parts.append(f" LIMIT {limit_param} OFFSET {offset_param}")
+        self.params.extend([limit, offset])
+        return self
+
+    def build(self) -> tuple[str, list[Any]]:
+        """
+        Build the final query and parameters.
+
+        Returns:
+            Tuple of (query_string, parameters)
+        """
+        return ("".join(self.query_parts), self.params)
+
+
+# AIDEV-NOTE: Helper function to safely format queries with dynamic sort fields
+@beartype
+def safe_format_query(
+    query_template: str,
+    *,
+    sort_by: str | None = None,
+    direction: str | None = None,
+    allowed_sort_fields: set[str] | None = None,
+    table_prefix: str = "",
+    **kwargs: Any,
+) -> str:
+    """
+    Safely format a query template with validated sort fields and direction.
+
+    Args:
+        query_template: The query template with {sort_by} and {direction} placeholders
+        sort_by: The field to sort by
+        direction: Sort direction
+        allowed_sort_fields: Allowed fields for sorting
+        table_prefix: Optional table prefix for sort field
+        **kwargs: Additional template parameters
+
+    Returns:
+        The formatted query string
+    """
+    format_params = kwargs.copy()
+
+    if sort_by is not None:
+        format_params["sort_by"] = validate_sort_field(
+            sort_by, allowed_sort_fields, table_prefix
+        )
+
+    if direction is not None:
+        format_params["direction"] = validate_sort_direction(direction)
+
+    return query_template.format(**format_params)
diff --git a/agents-api/docs/SQL_INJECTION_PREVENTION.md b/agents-api/docs/SQL_INJECTION_PREVENTION.md
new file mode 100644
index 000000000..e646d358e
--- /dev/null
+++ b/agents-api/docs/SQL_INJECTION_PREVENTION.md
@@ -0,0 +1,164 @@
+# SQL Injection Prevention in Agents API
+
+## Overview
+
+This document describes the SQL injection prevention mechanisms implemented in the agents-api to protect against malicious SQL injection attacks.
+
+## Vulnerabilities Identified
+
+During the security audit, the following SQL injection vulnerabilities were found:
+
+1. **Direct String Interpolation**: Several query files used f-strings or `.format()` to directly interpolate user-controlled values into SQL queries, particularly in:
+   - ORDER BY clauses (sort field and direction)
+   - Dynamic query building with metadata filters
+   - Table/column name references
+
+2. **Affected Files**:
+   - `queries/entries/list_entries.py`
+   - `queries/docs/list_docs.py`
+   - `queries/files/list_files.py`
+   - `queries/docs/bulk_delete_docs.py`
+   - `queries/agents/list_agents.py`
+
+## Solution Implemented
+
+### 1. SQL Utilities Module (`queries/sql_utils.py`)
+
+Created a comprehensive SQL utilities module with the following components:
+
+#### a. `sanitize_identifier()`
+- Validates SQL identifiers (table/column names)
+- Enforces alphanumeric + underscore pattern
+- Blocks SQL keywords
+- Enforces PostgreSQL 63-character limit
+
+#### b. `validate_sort_field()` and `validate_sort_direction()`
+- Whitelist-based validation for sort fields
+- Ensures sort directions are only "ASC" or "DESC"
+- Supports custom allowed field lists
+- Adds table prefixes safely
+
+#### c. `SafeQueryBuilder` Class
+- Provides a safe way to build dynamic SQL queries
+- All parameters are properly parameterized ($1, $2, etc.)
+- Supports complex query construction without string concatenation
+- Methods for adding conditions, ORDER BY, LIMIT/OFFSET safely
+
+#### d. `safe_format_query()`
+- Safe alternative to string formatting for query templates
+- Validates all dynamic parts before formatting
+- Primarily used for ORDER BY clauses
+
+### 2. Updated Query Files
+
+Modified vulnerable query files to use the new safety mechanisms:
+
+- **list_entries.py**: Uses `safe_format_query()` for ORDER BY
+- **list_docs.py**: Uses `SafeQueryBuilder` for entire query construction
+- **list_files.py**: Uses `SafeQueryBuilder` for dynamic conditions
+- **bulk_delete_docs.py**: Removed f-string usage, uses proper concatenation
+- **list_agents.py**: Uses `.replace()` instead of `.format()` for metadata filter
+
+### 3. Test Coverage
+
+Added comprehensive test suite (`tests/test_sql_injection_prevention.py`) that verifies:
+- Identifier sanitization
+- SQL injection attempt blocking
+- Whitelist validation
+- Safe query building
+- Complex query construction
+
+## Usage Examples
+
+### Basic Query Building
+```python
+from agents_api.queries.sql_utils import SafeQueryBuilder
+
+builder = SafeQueryBuilder("SELECT * FROM users WHERE active = true")
+builder.add_condition(" AND created_at > {}", "2024-01-01")
+builder.add_order_by("created_at", "desc")
+builder.add_limit_offset(10, 0)
+
+query, params = builder.build()
+# Result: SELECT * FROM users WHERE active = true AND created_at > $1 ORDER BY created_at DESC LIMIT $2 OFFSET $3
+# Params: ['2024-01-01', 10, 0]
+```
+
+### Safe ORDER BY Formatting
+```python
+from agents_api.queries.sql_utils import safe_format_query
+
+query = safe_format_query(
+    "SELECT * FROM entries ORDER BY {sort_by} {direction}",
+    sort_by="timestamp",
+    direction="desc",
+    allowed_sort_fields={"created_at", "timestamp"}
+)
+# Result: SELECT * FROM entries ORDER BY timestamp DESC
+```
+
+### Preventing SQL Injection
+```python
+# This will raise HTTPException(400)
+safe_format_query(
+    "SELECT * FROM users ORDER BY {sort_by}",
+    sort_by="created_at; DROP TABLE users;--"
+)
+```
+
+## Best Practices
+
+1. **Never use f-strings or .format() for SQL queries** with user input
+2. **Always use parameterized queries** ($1, $2, etc.) for values
+3. **Whitelist column/table names** that can be dynamically referenced
+4. **Use SafeQueryBuilder** for complex dynamic query construction
+5. **Validate all user input** before using in SQL queries
+
+## Migration Guide
+
+If you need to update other query files:
+
+1. **For simple ORDER BY**:
+   ```python
+   # Before
+   query = f"SELECT * FROM table ORDER BY {sort_by} {direction}"
+   
+   # After
+   from ..sql_utils import safe_format_query
+   query = safe_format_query(
+       "SELECT * FROM table ORDER BY {sort_by} {direction}",
+       sort_by=sort_by,
+       direction=direction,
+       allowed_sort_fields={"created_at", "updated_at"}
+   )
+   ```
+
+2. **For complex dynamic queries**:
+   ```python
+   # Before
+   query = base_query
+   if condition:
+       query += f" AND field = {value}"
+   
+   # After
+   from ..sql_utils import SafeQueryBuilder
+   builder = SafeQueryBuilder(base_query)
+   if condition:
+       builder.add_condition(" AND field = {}", value)
+   query, params = builder.build()
+   ```
+
+## Security Considerations
+
+- The whitelist approach ensures only pre-approved column names can be used
+- All values are parameterized, preventing SQL injection via data
+- SQL keywords and special characters in identifiers are blocked
+- The 63-character limit prevents buffer overflow attacks
+- HTTPException with 400 status provides clear error messages without exposing internals
+
+## Future Improvements
+
+1. Consider adding more SQL keywords to the blocklist
+2. Add support for more complex query patterns if needed
+3. Consider integrating with an SQL parser for even more robust validation
+4. Add logging for blocked SQL injection attempts for security monitoring
\ No newline at end of file
diff --git a/agents-api/examples/sql_injection_prevention_demo.py b/agents-api/examples/sql_injection_prevention_demo.py
new file mode 100644
index 000000000..81bf9ceec
--- /dev/null
+++ b/agents-api/examples/sql_injection_prevention_demo.py
@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""
+SQL Injection Prevention Demo
+
+This script demonstrates how the SQL injection prevention mechanisms work
+in the agents-api queries module.
+"""
+
+import sys
+
+sys.path.insert(0, ".")
+
+from agents_api.queries.sql_utils import (
+    SafeQueryBuilder,
+    safe_format_query,
+    validate_sort_field,
+)
+
+
+def demonstrate_safe_query_building():
+    """Demonstrate safe query building with SafeQueryBuilder."""
+    print("=== SafeQueryBuilder Demo ===\n")
+
+    # Example 1: Basic safe query construction
+    print("1. Basic safe query construction:")
+    builder = SafeQueryBuilder("SELECT * FROM agents WHERE developer_id = $1", ["dev-123"])
+    builder.add_condition(" AND name LIKE {}", "%test%")
+    builder.add_condition(" AND created_at > {}", "2024-01-01")
+    builder.add_order_by("created_at", "desc")
+    builder.add_limit_offset(10, 0)
+
+    query, params = builder.build()
+    print(f"Query: {query}")
+    print(f"Params: {params}")
+    print()
+
+
+def demonstrate_sql_injection_prevention():
+    """Demonstrate how SQL injection attempts are prevented."""
+    print("=== SQL Injection Prevention Demo ===\n")
+
+    # Example 2: Preventing SQL injection in ORDER BY
+    print("2. Attempting SQL injection in ORDER BY clause:")
+    malicious_sort = "created_at; DROP TABLE agents;--"
+
+    try:
+        result = safe_format_query(
+            "SELECT * FROM agents ORDER BY {sort_by} {direction}",
+            sort_by=malicious_sort,
+            direction="desc",
+        )
+        print(f"Result: {result}")
+    except Exception as e:
+        print(f"✓ Blocked: {type(e).__name__}: {e}")
+    print()
+
+    # Example 3: Preventing SQL injection in sort direction
+    print("3. Attempting SQL injection in sort direction:")
+    malicious_direction = "desc; DELETE FROM agents WHERE 1=1;--"
+
+    try:
+        result = safe_format_query(
+            "SELECT * FROM agents ORDER BY {sort_by} {direction}",
+            sort_by="created_at",
+            direction=malicious_direction,
+        )
+        print(f"Result: {result}")
+    except Exception as e:
+        print(f"✓ Blocked: {type(e).__name__}: {e}")
+    print()
+
+
+def demonstrate_valid_usage():
+    """Demonstrate valid usage of the SQL utilities."""
+    print("=== Valid Usage Demo ===\n")
+
+    # Example 4: Valid query formatting
+    print("4. Valid query formatting with safe_format_query:")
+    query = safe_format_query(
+        "SELECT * FROM documents WHERE owner_id = $1 ORDER BY {sort_by} {direction}",
+        sort_by="updated_at",
+        direction="asc",
+        allowed_sort_fields={"created_at", "updated_at", "title"},
+    )
+    print(f"Safe query: {query}")
+    print()
+
+    # Example 5: Complex query with metadata filters
+    print("5. Complex query with SafeQueryBuilder:")
+    builder = SafeQueryBuilder(
+        """
+        SELECT d.*, array_agg(t.tag) as tags
+        FROM documents d
+        LEFT JOIN document_tags dt ON d.id = dt.document_id
+        LEFT JOIN tags t ON dt.tag_id = t.id
+        WHERE d.developer_id = $1
+    """,
+        ["dev-456"],
+    )
+
+    builder.add_condition(" AND d.status = {}", "published")
+    builder.add_condition(" AND d.created_at BETWEEN {} AND {}", "2024-01-01", "2024-12-31")
+    builder.add_raw_condition(" GROUP BY d.id")
+    builder.add_order_by(
+        "created_at", "desc", allowed_fields={"created_at", "updated_at"}, table_prefix="d."
+    )
+    builder.add_limit_offset(20, 0)
+
+    query, params = builder.build()
+    print(f"Query: {query}")
+    print(f"Params: {params}")
+    print()
+
+
+def demonstrate_whitelist_validation():
+    """Demonstrate whitelist-based validation."""
+    print("=== Whitelist Validation Demo ===\n")
+
+    # Example 6: Field validation with custom whitelist
+    print("6. Field validation with custom whitelist:")
+
+    # Valid field
+    try:
+        field = validate_sort_field(
+            "published_at", allowed_fields={"published_at", "author_name", "view_count"}
+        )
+        print(f"✓ Valid field accepted: {field}")
+    except Exception as e:
+        print(f"✗ Error: {e}")
+
+    # Invalid field
+    try:
+        field = validate_sort_field(
+            "secret_data", allowed_fields={"published_at", "author_name", "view_count"}
+        )
+        print(f"Field accepted: {field}")
+    except Exception as e:
+        print(f"✓ Invalid field blocked: {type(e).__name__}: {e}")
+    print()
+
+
+if __name__ == "__main__":
+    print("SQL Injection Prevention Demonstration")
+    print("=====================================\n")
+
+    demonstrate_safe_query_building()
+    demonstrate_sql_injection_prevention()
+    demonstrate_valid_usage()
+    demonstrate_whitelist_validation()
+
+    print("\nConclusion:")
+    print("-----------")
+    print("The SQL injection prevention mechanisms ensure that:")
+    print("1. All dynamic SQL parts are properly validated")
+    print("2. User input is parameterized, not concatenated")
+    print("3. Sort fields and directions are whitelisted")
+    print("4. SQL keywords and special characters are blocked")
+    print("5. Complex queries can be built safely without string concatenation")
diff --git a/agents-api/tests/test_sql_injection_prevention.py b/agents-api/tests/test_sql_injection_prevention.py
new file mode 100644
index 000000000..3cce6423a
--- /dev/null
+++ b/agents-api/tests/test_sql_injection_prevention.py
@@ -0,0 +1,260 @@
+"""
+Tests for SQL injection prevention mechanisms.
+
+This module tests the SQL utilities and query builders to ensure they properly
+prevent SQL injection attacks through various attack vectors.
+"""
+
+from agents_api.queries.sql_utils import (
+    SafeQueryBuilder,
+    safe_format_query,
+    sanitize_identifier,
+    validate_sort_direction,
+    validate_sort_field,
+)
+from fastapi import HTTPException
+from ward import raises, test
+
+
+@test("sanitize_identifier: valid identifiers pass through unchanged")
+def test_valid_identifiers():
+    assert sanitize_identifier("column_name") == "column_name"
+    assert sanitize_identifier("_private_column") == "_private_column"
+    assert sanitize_identifier("column123") == "column123"
+    assert sanitize_identifier("CamelCase") == "CamelCase"
+
+
+@test("sanitize_identifier: empty string raises HTTPException")
+def test_empty_identifier():
+    with raises(HTTPException) as exc_info:
+        sanitize_identifier("")
+    assert exc_info.raised.status_code == 400
+    assert "cannot be empty" in exc_info.raised.detail
+
+
+@test("sanitize_identifier: identifiers starting with numbers are rejected")
+def test_identifier_starting_with_number():
+    with raises(HTTPException) as exc_info:
+        sanitize_identifier("123column")
+    assert exc_info.raised.status_code == 400
+
+
+@test("sanitize_identifier: identifiers with spaces are rejected")
+def test_identifier_with_spaces():
+    with raises(HTTPException) as exc_info:
+        sanitize_identifier("column name")
+    assert exc_info.raised.status_code == 400
+
+
+@test("sanitize_identifier: identifiers with special chars are rejected")
+def test_identifier_with_special_chars():
+    with raises(HTTPException) as exc_info:
+        sanitize_identifier("column-name")
+    assert exc_info.raised.status_code == 400
+
+
+@test("sanitize_identifier: SQL injection attempts are blocked")
+def test_sql_injection_attempts():
+    with raises(HTTPException):
+        sanitize_identifier("column; DROP TABLE users;--")
+
+    with raises(HTTPException):
+        sanitize_identifier("column' OR '1'='1")
+
+
+@test("sanitize_identifier: SQL keywords are rejected")
+def test_sql_keywords():
+    keywords = ["select", "SELECT", "drop", "DROP", "table", "TABLE"]
+    for keyword in keywords:
+        with raises(HTTPException) as exc_info:
+            sanitize_identifier(keyword)
+        assert exc_info.raised.status_code == 400
+        assert "reserved SQL keyword" in exc_info.raised.detail
+
+
+@test("sanitize_identifier: identifiers exceeding 63 chars are rejected")
+def test_length_limit():
+    long_name = "a" * 64  # PostgreSQL limit is 63 characters
+    with raises(HTTPException) as exc_info:
+        sanitize_identifier(long_name)
+    assert exc_info.raised.status_code == 400
+    assert "too long" in exc_info.raised.detail
+
+
+@test("validate_sort_field: allowed fields pass validation")
+def test_allowed_sort_fields():
+    assert validate_sort_field("created_at") == "created_at"
+    assert validate_sort_field("updated_at") == "updated_at"
+    assert validate_sort_field("timestamp") == "timestamp"
+
+
+@test("validate_sort_field: custom allowed fields work correctly")
+def test_custom_allowed_fields():
+    custom_fields = {"custom_field", "another_field"}
+    assert validate_sort_field("custom_field", custom_fields) == "custom_field"
+
+
+@test("validate_sort_field: table prefixes are properly added")
+def test_table_prefix():
+    assert validate_sort_field("created_at", table_prefix="t.") == "t.created_at"
+    assert validate_sort_field("updated_at", table_prefix="users.") == "users.updated_at"
+
+
+@test("validate_sort_field: invalid fields raise HTTPException")
+def test_invalid_sort_fields():
+    with raises(HTTPException) as exc_info:
+        validate_sort_field("invalid_field")
+    assert exc_info.raised.status_code == 400
+    assert "Invalid sort field" in exc_info.raised.detail
+
+
+@test("validate_sort_field: SQL injection in field names is blocked")
+def test_sort_field_sql_injection():
+    with raises(HTTPException):
+        validate_sort_field("created_at; DROP TABLE users;--")
+
+
+@test("validate_sort_direction: valid directions are normalized")
+def test_valid_sort_directions():
+    assert validate_sort_direction("asc") == "ASC"
+    assert validate_sort_direction("ASC") == "ASC"
+    assert validate_sort_direction("desc") == "DESC"
+    assert validate_sort_direction("DESC") == "DESC"
+
+
+@test("validate_sort_direction: invalid directions raise HTTPException")
+def test_invalid_sort_directions():
+    invalid_directions = ["ascending", "descending", "up", "down", "", "'; DROP TABLE--"]
+    for direction in invalid_directions:
+        with raises(HTTPException) as exc_info:
+            validate_sort_direction(direction)
+        assert exc_info.raised.status_code == 400
+        assert "Invalid sort direction" in exc_info.raised.detail
+
+
+@test("SafeQueryBuilder: basic query construction works")
+def test_basic_query_building():
+    builder = SafeQueryBuilder("SELECT * FROM users WHERE 1=1")
+    builder.add_condition(" AND name = {}", "John")
+    builder.add_condition(" AND age > {}", 25)
+
+    query, params = builder.build()
+    assert query == "SELECT * FROM users WHERE 1=1 AND name = $1 AND age > $2"
+    assert params == ["John", 25]
+
+
+@test("SafeQueryBuilder: works with initial parameters")
+def test_query_with_initial_params():
+    builder = SafeQueryBuilder("SELECT * FROM users WHERE company_id = $1", [123])
+    builder.add_condition(" AND active = {}", True)
+
+    query, params = builder.build()
+    assert query == "SELECT * FROM users WHERE company_id = $1 AND active = $2"
+    assert params == [123, True]
+
+
+@test("SafeQueryBuilder: ORDER BY clause with validation")
+def test_order_by_clause():
+    builder = SafeQueryBuilder("SELECT * FROM users")
+    builder.add_order_by("created_at", "desc")
+
+    query, _ = builder.build()
+    assert "ORDER BY created_at DESC" in query
+
+
+@test("SafeQueryBuilder: ORDER BY with custom fields and table prefix")
+def test_order_by_custom_fields():
+    builder = SafeQueryBuilder("SELECT * FROM posts")
+    builder.add_order_by(
+        "published_at", "asc", allowed_fields={"published_at", "title"}, table_prefix="p."
+    )
+
+    query, _ = builder.build()
+    assert "ORDER BY p.published_at ASC" in query
+
+
+@test("SafeQueryBuilder: LIMIT and OFFSET clauses")
+def test_limit_offset():
+    builder = SafeQueryBuilder("SELECT * FROM users")
+    builder.add_limit_offset(10, 20)
+
+    query, params = builder.build()
+    assert "LIMIT $1 OFFSET $2" in query
+    assert params == [10, 20]
+
+
+@test("SafeQueryBuilder: complex query with multiple operations")
+def test_complex_query():
+    builder = SafeQueryBuilder("SELECT * FROM documents WHERE developer_id = $1", ["uuid-123"])
+    builder.add_condition(" AND status = {}", "active")
+    builder.add_condition(" AND created_at > {}", "2024-01-01")
+    builder.add_order_by("created_at", "desc")
+    builder.add_limit_offset(50, 0)
+
+    query, params = builder.build()
+    expected = (
+        "SELECT * FROM documents WHERE developer_id = $1"
+        " AND status = $2"
+        " AND created_at > $3"
+        " ORDER BY created_at DESC"
+        " LIMIT $4 OFFSET $5"
+    )
+    assert query == expected
+    assert params == ["uuid-123", "active", "2024-01-01", 50, 0]
+
+
+@test("safe_format_query: basic query template formatting")
+def test_basic_formatting():
+    template = "SELECT * FROM users ORDER BY {sort_by} {direction}"
+    result = safe_format_query(template, sort_by="created_at", direction="desc")
+    assert result == "SELECT * FROM users ORDER BY created_at DESC"
+
+
+@test("safe_format_query: formatting with table prefix")
+def test_format_with_table_prefix():
+    template = "SELECT * FROM users u ORDER BY {sort_by} {direction}"
+    result = safe_format_query(
+        template, sort_by="updated_at", direction="asc", table_prefix="u."
+    )
+    assert result == "SELECT * FROM users u ORDER BY u.updated_at ASC"
+
+
+@test("safe_format_query: SQL injection via sort_by is prevented")
+def test_format_sql_injection_sort_by():
+    template = "SELECT * FROM users ORDER BY {sort_by} {direction}"
+
+    with raises(HTTPException):
+        safe_format_query(template, sort_by="created_at; DROP TABLE users;--", direction="asc")
+
+
+@test("safe_format_query: SQL injection via direction is prevented")
+def test_format_sql_injection_direction():
+    template = "SELECT * FROM users ORDER BY {sort_by} {direction}"
+
+    with raises(HTTPException):
+        safe_format_query(template, sort_by="created_at", direction="asc; DROP TABLE users;--")
+
+
+@test("safe_format_query: custom allowed fields work correctly")
+def test_format_custom_allowed_fields():
+    template = "SELECT * FROM posts ORDER BY {sort_by} {direction}"
+    result = safe_format_query(
+        template,
+        sort_by="published_at",
+        direction="desc",
+        allowed_sort_fields={"published_at", "author_id", "title"},
+    )
+    assert result == "SELECT * FROM posts ORDER BY published_at DESC"
+
+
+@test("safe_format_query: non-allowed fields are rejected")
+def test_format_non_allowed_field():
+    template = "SELECT * FROM posts ORDER BY {sort_by} {direction}"
+
+    with raises(HTTPException):
+        safe_format_query(
+            template,
+            sort_by="created_at",  # Not in allowed_sort_fields
+            direction="desc",
+            allowed_sort_fields={"published_at", "author_id", "title"},
+        )