diff --git a/modules/connectors/connectorTicketJira.py b/modules/connectors/connectorTicketJira.py
new file mode 100644
index 00000000..f2edd200
--- /dev/null
+++ b/modules/connectors/connectorTicketJira.py
@@ -0,0 +1,240 @@
+"""Jira connector for CRUD operations."""
+
+from dataclasses import dataclass
+import logging
+import aiohttp
+import json
+from typing import Optional
+
+from modules.interfaces.interfaceTicketModel import (
+    TicketBase,
+    TicketFieldAttribute,
+    Task,
+)
+
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ConnectorTicketJira(TicketBase):
+    jira_username: str
+    jira_api_token: str
+    jira_url: str
+    project_code: str
+    issue_type: str
+
+    @classmethod
+    async def create(
+        cls,
+        *,
+        jira_username: str,
+        jira_api_token: str,
+        jira_url: str,
+        project_code: str,
+        issue_type: str,
+    ):
+        return ConnectorTicketJira(
+            jira_username=jira_username,
+            jira_api_token=jira_api_token,
+            jira_url=jira_url,
+            project_code=project_code,
+            issue_type=issue_type,
+        )
+
+    async def read_attributes(self) -> list[TicketFieldAttribute]:
+        """
+        Read field attributes from Jira by querying for a single issue
+        and extracting the field mappings.
+
+        Returns:
+            list[TicketFieldAttribute]: List of field attributes with names and IDs
+        """
+        jql_query = f"project={self.project_code} AND issuetype={self.issue_type}"
+
+        # Prepare the request URL and parameters
+        url = f"{self.jira_url}/rest/api/2/search"
+        params = {"jql": jql_query, "maxResults": 1, "expand": "names"}
+
+        # Prepare authentication
+        auth = aiohttp.BasicAuth(self.jira_username, self.jira_api_token)
+
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.get(url, params=params, auth=auth) as response:
+                    if response.status != 200:
+                        error_text = await response.text()
+                        logger.error(
+                            f"Jira API request failed with status {response.status}: {error_text}"
+                        )
+                        raise Exception(
+                            f"Jira API request failed with status {response.status}"
+                        )
+
+                    data = await response.json()
+
+                    # Extract issues and field names
+                    issues = data.get("issues", [])
+                    field_names = data.get("names", {})
+
+                    if not issues:
+                        logger.warning(f"No issues found for query: {jql_query}")
+                        return []
+
+                    # Extract field attributes from the first issue
+                    attributes = []
+                    issue = issues[0]
+                    fields = issue.get("fields", {})
+
+                    for field_id, value in fields.items():
+                        field_name = field_names.get(field_id, field_id)
+                        attributes.append(
+                            TicketFieldAttribute(field_name=field_name, field=field_id)
+                        )
+
+                    logger.info(
+                        f"Successfully retrieved {len(attributes)} field attributes from Jira"
+                    )
+                    return attributes
+
+        except aiohttp.ClientError as e:
+            logger.error(f"HTTP client error while fetching Jira attributes: {str(e)}")
+            raise Exception(f"Failed to connect to Jira: {str(e)}")
+        except json.JSONDecodeError as e:
+            logger.error(f"Failed to parse Jira API response: {str(e)}")
+            raise Exception(f"Invalid response from Jira API: {str(e)}")
+        except Exception as e:
+            logger.error(f"Unexpected error while fetching Jira attributes: {str(e)}")
+            raise
+
+    async def read_tasks(self, *, limit: int = 0) -> list[Task]:
+        """
+        Read tasks from Jira with pagination support.
+
+        Args:
+            limit: Maximum number of tasks to retrieve. 0 means no limit.
+
+        Returns:
+            list[Task]: List of tasks with their data
+        """
+        jql_query = f"project={self.project_code} AND issuetype={self.issue_type}"
+
+        # Initialize variables for pagination
+        start_at = 0
+        max_results = 50
+        total = 1  # Initialize with a value greater than 0 to enter the loop
+        tasks = []
+
+        # Prepare authentication
+        auth = aiohttp.BasicAuth(self.jira_username, self.jira_api_token)
+        url = f"{self.jira_url}/rest/api/2/search"
+
+        try:
+            async with aiohttp.ClientSession() as session:
+                while start_at < total and (limit == 0 or len(tasks) < limit):
+                    # Prepare request parameters
+                    params = {
+                        "jql": jql_query,
+                        "startAt": start_at,
+                        "maxResults": max_results,
+                    }
+
+                    headers = {"Content-Type": "application/json"}
+
+                    async with session.get(
+                        url, params=params, auth=auth, headers=headers
+                    ) as response:
+                        if response.status != 200:
+                            error_text = await response.text()
+                            logger.error(
+                                f"Failed to fetch tasks from Jira. Status code: {response.status}, Response: {error_text}"
+                            )
+                            break
+
+                        data = await response.json()
+                        issues = data.get("issues", [])
+                        total = data.get("total", 0)
+
+                        for issue in issues:
+                            # Create task with all issue data
+                            task_data = {
+                                "id": issue.get("id"),
+                                "key": issue.get("key"),
+                                "fields": issue.get("fields", {}),
+                                "self": issue.get("self"),
+                                "expand": issue.get("expand", ""),
+                            }
+
+                            task = Task(data=task_data)
+                            tasks.append(task)
+
+                            # Check limit
+                            if limit > 0 and len(tasks) >= limit:
+                                break
+
+                        start_at += max_results
+                        logger.debug(f"Issues packages reading: {len(tasks)}")
+
+                logger.info(f"JIRA issues read: {len(tasks)}")
+                return tasks
+
+        except aiohttp.ClientError as e:
+            logger.error(f"HTTP client error while fetching Jira tasks: {str(e)}")
+            raise Exception(f"Failed to connect to Jira: {str(e)}")
+        except json.JSONDecodeError as e:
+            logger.error(f"Failed to parse Jira API response: {str(e)}")
+            raise Exception(f"Invalid response from Jira API: {str(e)}")
+        except Exception as e:
+            logger.error(f"Unexpected error while fetching Jira tasks: {str(e)}")
+            raise
+
+    async def write_tasks(self, tasklist: list[Task]) -> None:
+        """
+        Write/update tasks to Jira.
+
+        Args:
+            tasklist: List of Task objects containing task data to update
+        """
+        headers = {"Accept": "application/json", "Content-Type": "application/json"}
+        auth = aiohttp.BasicAuth(self.jira_username, self.jira_api_token)
+
+        try:
+            async with aiohttp.ClientSession() as session:
+                for task in tasklist:
+                    task_data = task.data
+                    task_id = task_data.get("id") or task_data.get("key")
+
+                    if not task_id:
+                        logger.warning("Task missing ID or key, skipping update")
+                        continue
+
+                    # Extract fields to update from task data
+                    fields = task_data.get("fields", {})
+
+                    if not fields:
+                        logger.debug(f"No fields to update for task {task_id}")
+                        continue
+
+                    # Prepare update data
+                    update_data = {"fields": fields}
+
+                    # Make the update request
+                    url = f"{self.jira_url}/rest/api/2/issue/{task_id}"
+
+                    async with session.put(
+                        url, json=update_data, headers=headers, auth=auth
+                    ) as response:
+                        if response.status == 204:
+                            logger.info(f"JIRA task {task_id} updated successfully.")
+                        else:
+                            error_text = await response.text()
+                            logger.error(
+                                f"JIRA failed to update task {task_id}: {response.status} - {error_text}"
+                            )
+
+        except aiohttp.ClientError as e:
+            logger.error(f"HTTP client error while updating Jira tasks: {str(e)}")
+            raise Exception(f"Failed to connect to Jira: {str(e)}")
+        except Exception as e:
+            logger.error(f"Unexpected error while updating Jira tasks: {str(e)}")
+            raise
diff --git a/modules/interfaces/interfaceTicketModel.py b/modules/interfaces/interfaceTicketModel.py
new file mode 100644
index 00000000..151eabac
--- /dev/null
+++ b/modules/interfaces/interfaceTicketModel.py
@@ -0,0 +1,26 @@
+"""Base class for ticket classes."""
+
+from typing import Any, Dict
+from pydantic import BaseModel, Field
+from abc import ABC, abstractmethod
+
+
+class TicketFieldAttribute(BaseModel):
+    field_name: str = Field(description="Human-readable field name")
+    field: str = Field(description="JIRA field ID/key")
+
+
+class Task(BaseModel):
+    # A very flexible approach for now. Might want to be more strict in the future.
+    data: Dict[str, Any] = Field(default_factory=dict, description="Task data")
+
+
+class TicketBase(ABC):
+    @abstractmethod
+    async def read_attributes(self) -> list[TicketFieldAttribute]: ...
+
+    @abstractmethod
+    async def read_tasks(self) -> list[Task]: ...
+
+    @abstractmethod
+    async def write_tasks(self, tasklist: list[Task]) -> None: ...
diff --git a/modules/interfaces/interfaceTicketObjects.py b/modules/interfaces/interfaceTicketObjects.py
new file mode 100644
index 00000000..e1ac75a8
--- /dev/null
+++ b/modules/interfaces/interfaceTicketObjects.py
@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+
+
+SUPPORTED_SYSTEMS = ["jira"]
+
+
+@dataclass(slots=True)
+class TicketInterface:
+    # TODO: user must create instance of Ticket connector
+    ticketConnector = None