Merge feat/2: Implement MCP server core and authentication

pyproject.toml

@@ -35,6 +35,9 @@ dev = [
     "pytest-asyncio>=0.21.0",
 ]
 
+[project.scripts]
+gitea-mcp = "gitea_mcp.server:main"
+
 [project.urls]
 Homepage = "https://github.com/lmiranda/gitea-mcp-remote"
 Repository = "https://github.com/lmiranda/gitea-mcp-remote"

src/gitea_mcp/auth.py

@@ -0,0 +1,51 @@
+"""Authentication and configuration management for Gitea MCP server."""
+
+import os
+from typing import Optional
+from dotenv import load_dotenv
+
+
+class AuthConfig:
+    """Manages authentication configuration for Gitea API."""
+
+    def __init__(self):
+        """Initialize authentication configuration from environment variables."""
+        load_dotenv()
+
+        self.api_url: Optional[str] = os.getenv("GITEA_API_URL")
+        self.api_token: Optional[str] = os.getenv("GITEA_API_TOKEN")
+
+        self._validate()
+
+    def _validate(self) -> None:
+        """Validate that required configuration is present.
+
+        Raises:
+            ValueError: If required environment variables are missing.
+        """
+        if not self.api_url:
+            raise ValueError(
+                "GITEA_API_URL environment variable is required. "
+                "Please set it in your .env file or environment."
+            )
+
+        if not self.api_token:
+            raise ValueError(
+                "GITEA_API_TOKEN environment variable is required. "
+                "Please set it in your .env file or environment."
+            )
+
+        # Remove trailing slash from URL if present
+        if self.api_url.endswith("/"):
+            self.api_url = self.api_url[:-1]
+
+    def get_auth_headers(self) -> dict[str, str]:
+        """Get authentication headers for API requests.
+
+        Returns:
+            dict: HTTP headers with authorization token.
+        """
+        return {
+            "Authorization": f"token {self.api_token}",
+            "Content-Type": "application/json",
+        }

src/gitea_mcp/client.py

@@ -0,0 +1,216 @@
+"""HTTP client for Gitea API."""
+
+import httpx
+from typing import Any, Optional
+from .auth import AuthConfig
+
+
+class GiteaClientError(Exception):
+    """Base exception for Gitea client errors."""
+
+    pass
+
+
+class GiteaAuthError(GiteaClientError):
+    """Authentication error."""
+
+    pass
+
+
+class GiteaNotFoundError(GiteaClientError):
+    """Resource not found error."""
+
+    pass
+
+
+class GiteaServerError(GiteaClientError):
+    """Server error."""
+
+    pass
+
+
+class GiteaClient:
+    """Async HTTP client for Gitea API."""
+
+    def __init__(self, config: AuthConfig, timeout: float = 30.0):
+        """Initialize Gitea API client.
+
+        Args:
+            config: Authentication configuration.
+            timeout: Request timeout in seconds (default: 30.0).
+        """
+        self.config = config
+        self.timeout = timeout
+        self._client: Optional[httpx.AsyncClient] = None
+
+    async def __aenter__(self):
+        """Async context manager entry."""
+        self._client = httpx.AsyncClient(
+            base_url=self.config.api_url,
+            headers=self.config.get_auth_headers(),
+            timeout=self.timeout,
+        )
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        """Async context manager exit."""
+        if self._client:
+            await self._client.aclose()
+
+    def _handle_error(self, response: httpx.Response) -> None:
+        """Handle HTTP error responses.
+
+        Args:
+            response: HTTP response object.
+
+        Raises:
+            GiteaAuthError: For 401/403 errors.
+            GiteaNotFoundError: For 404 errors.
+            GiteaServerError: For 500+ errors.
+            GiteaClientError: For other errors.
+        """
+        status = response.status_code
+
+        if status == 401:
+            raise GiteaAuthError(
+                "Authentication failed. Please check your GITEA_API_TOKEN."
+            )
+        elif status == 403:
+            raise GiteaAuthError(
+                "Access forbidden. Your API token may not have required permissions."
+            )
+        elif status == 404:
+            raise GiteaNotFoundError(
+                f"Resource not found: {response.request.url}"
+            )
+        elif status >= 500:
+            raise GiteaServerError(
+                f"Gitea server error (HTTP {status}): {response.text}"
+            )
+        else:
+            raise GiteaClientError(
+                f"API request failed (HTTP {status}): {response.text}"
+            )
+
+    async def get(self, path: str, **kwargs) -> dict[str, Any]:
+        """Make GET request to Gitea API.
+
+        Args:
+            path: API endpoint path (e.g., "/api/v1/repos/owner/repo").
+            **kwargs: Additional arguments for httpx request.
+
+        Returns:
+            dict: JSON response data.
+
+        Raises:
+            GiteaClientError: If request fails.
+        """
+        if not self._client:
+            raise GiteaClientError(
+                "Client not initialized. Use 'async with' context manager."
+            )
+
+        try:
+            response = await self._client.get(path, **kwargs)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError:
+            self._handle_error(response)
+        except httpx.RequestError as e:
+            raise GiteaClientError(
+                f"Request failed: {e}"
+            ) from e
+
+    async def post(self, path: str, json: Optional[dict[str, Any]] = None, **kwargs) -> dict[str, Any]:
+        """Make POST request to Gitea API.
+
+        Args:
+            path: API endpoint path.
+            json: JSON data to send in request body.
+            **kwargs: Additional arguments for httpx request.
+
+        Returns:
+            dict: JSON response data.
+
+        Raises:
+            GiteaClientError: If request fails.
+        """
+        if not self._client:
+            raise GiteaClientError(
+                "Client not initialized. Use 'async with' context manager."
+            )
+
+        try:
+            response = await self._client.post(path, json=json, **kwargs)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError:
+            self._handle_error(response)
+        except httpx.RequestError as e:
+            raise GiteaClientError(
+                f"Request failed: {e}"
+            ) from e
+
+    async def patch(self, path: str, json: Optional[dict[str, Any]] = None, **kwargs) -> dict[str, Any]:
+        """Make PATCH request to Gitea API.
+
+        Args:
+            path: API endpoint path.
+            json: JSON data to send in request body.
+            **kwargs: Additional arguments for httpx request.
+
+        Returns:
+            dict: JSON response data.
+
+        Raises:
+            GiteaClientError: If request fails.
+        """
+        if not self._client:
+            raise GiteaClientError(
+                "Client not initialized. Use 'async with' context manager."
+            )
+
+        try:
+            response = await self._client.patch(path, json=json, **kwargs)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError:
+            self._handle_error(response)
+        except httpx.RequestError as e:
+            raise GiteaClientError(
+                f"Request failed: {e}"
+            ) from e
+
+    async def delete(self, path: str, **kwargs) -> Optional[dict[str, Any]]:
+        """Make DELETE request to Gitea API.
+
+        Args:
+            path: API endpoint path.
+            **kwargs: Additional arguments for httpx request.
+
+        Returns:
+            dict or None: JSON response data if available, None for 204 responses.
+
+        Raises:
+            GiteaClientError: If request fails.
+        """
+        if not self._client:
+            raise GiteaClientError(
+                "Client not initialized. Use 'async with' context manager."
+            )
+
+        try:
+            response = await self._client.delete(path, **kwargs)
+            response.raise_for_status()
+
+            # DELETE requests may return 204 No Content
+            if response.status_code == 204:
+                return None
+
+            return response.json()
+        except httpx.HTTPStatusError:
+            self._handle_error(response)
+        except httpx.RequestError as e:
+            raise GiteaClientError(
+                f"Request failed: {e}"
+            ) from e

src/gitea_mcp/server.py

@@ -1,7 +1,166 @@
-"""MCP server implementation for Gitea API integration.
+"""MCP server implementation for Gitea API integration."""
 
-This module will contain the main MCP server implementation.
+import asyncio
+import argparse
+from mcp.server import Server
+from mcp.server.stdio import stdio_server
+from mcp.types import Tool, TextContent
+
+from . import __version__
+from .auth import AuthConfig
+from .client import GiteaClient, GiteaClientError
+
+
+# Global client instance
+gitea_client: GiteaClient | None = None
+
+
+async def serve() -> None:
+    """Run the MCP server."""
+    server = Server("gitea-mcp")
+
+    # Initialize authentication config
+    try:
+        config = AuthConfig()
+    except ValueError as e:
+        print(f"Configuration error: {e}")
+        raise
+
+    # Initialize Gitea client
+    global gitea_client
+    gitea_client = GiteaClient(config)
+
+    @server.list_tools()
+    async def list_tools() -> list[Tool]:
+        """List available MCP tools.
+
+        Returns:
+            list: Available tools (placeholder for future implementation).
         """
+        # Placeholder tools - will be implemented in issues #3, #4, #5
+        return [
+            Tool(
+                name="list_repositories",
+                description="List repositories in an organization (coming soon)",
+                inputSchema={
+                    "type": "object",
+                    "properties": {
+                        "org": {
+                            "type": "string",
+                            "description": "Organization name",
+                        }
+                    },
+                    "required": ["org"],
+                },
+            ),
+            Tool(
+                name="create_issue",
+                description="Create a new issue in a repository (coming soon)",
+                inputSchema={
+                    "type": "object",
+                    "properties": {
+                        "owner": {
+                            "type": "string",
+                            "description": "Repository owner",
+                        },
+                        "repo": {
+                            "type": "string",
+                            "description": "Repository name",
+                        },
+                        "title": {
+                            "type": "string",
+                            "description": "Issue title",
+                        },
+                        "body": {
+                            "type": "string",
+                            "description": "Issue body",
+                        },
+                    },
+                    "required": ["owner", "repo", "title"],
+                },
+            ),
+            Tool(
+                name="create_pull_request",
+                description="Create a new pull request (coming soon)",
+                inputSchema={
+                    "type": "object",
+                    "properties": {
+                        "owner": {
+                            "type": "string",
+                            "description": "Repository owner",
+                        },
+                        "repo": {
+                            "type": "string",
+                            "description": "Repository name",
+                        },
+                        "title": {
+                            "type": "string",
+                            "description": "Pull request title",
+                        },
+                        "head": {
+                            "type": "string",
+                            "description": "Source branch",
+                        },
+                        "base": {
+                            "type": "string",
+                            "description": "Target branch",
+                        },
+                    },
+                    "required": ["owner", "repo", "title", "head", "base"],
+                },
+            ),
+        ]
 
-# Placeholder for MCP server implementation
+    @server.call_tool()
-# TODO: Implement MCP server in future issues
+    async def call_tool(name: str, arguments: dict) -> list[TextContent]:
+        """Handle tool calls.
+
+        Args:
+            name: Tool name.
+            arguments: Tool arguments.
+
+        Returns:
+            list: Tool response.
+        """
+        # Placeholder implementation - actual tools will be implemented in future issues
+        return [
+            TextContent(
+                type="text",
+                text=f"Tool '{name}' is not yet implemented. Coming soon in issues #3, #4, #5.",
+            )
+        ]
+
+    # Run the server using stdio transport
+    async with stdio_server() as (read_stream, write_stream):
+        await server.run(
+            read_stream,
+            write_stream,
+            server.create_initialization_options(),
+        )
+
+
+def main() -> None:
+    """Main entry point with CLI argument parsing."""
+    parser = argparse.ArgumentParser(
+        description="Gitea MCP Server - MCP server for Gitea API integration"
+    )
+    parser.add_argument(
+        "--version",
+        action="version",
+        version=f"gitea-mcp {__version__}",
+    )
+
+    args = parser.parse_args()
+
+    # Run the server
+    try:
+        asyncio.run(serve())
+    except KeyboardInterrupt:
+        print("\nServer stopped by user")
+    except Exception as e:
+        print(f"Server error: {e}")
+        raise
+
+
+if __name__ == "__main__":
+    main()

tests/test_auth.py

@@ -0,0 +1,56 @@
+"""Tests for authentication module."""
+
+import os
+import pytest
+from gitea_mcp.auth import AuthConfig
+
+
+def test_auth_config_success(monkeypatch):
+    """Test successful authentication configuration."""
+    monkeypatch.setenv("GITEA_API_URL", "http://gitea.example.com/api/v1")
+    monkeypatch.setenv("GITEA_API_TOKEN", "test_token_123")
+
+    config = AuthConfig()
+
+    assert config.api_url == "http://gitea.example.com/api/v1"
+    assert config.api_token == "test_token_123"
+
+
+def test_auth_config_removes_trailing_slash(monkeypatch):
+    """Test that trailing slash is removed from URL."""
+    monkeypatch.setenv("GITEA_API_URL", "http://gitea.example.com/api/v1/")
+    monkeypatch.setenv("GITEA_API_TOKEN", "test_token_123")
+
+    config = AuthConfig()
+
+    assert config.api_url == "http://gitea.example.com/api/v1"
+
+
+def test_auth_config_missing_url(monkeypatch):
+    """Test error when GITEA_API_URL is missing."""
+    monkeypatch.delenv("GITEA_API_URL", raising=False)
+    monkeypatch.setenv("GITEA_API_TOKEN", "test_token_123")
+
+    with pytest.raises(ValueError, match="GITEA_API_URL"):
+        AuthConfig()
+
+
+def test_auth_config_missing_token(monkeypatch):
+    """Test error when GITEA_API_TOKEN is missing."""
+    monkeypatch.setenv("GITEA_API_URL", "http://gitea.example.com/api/v1")
+    monkeypatch.delenv("GITEA_API_TOKEN", raising=False)
+
+    with pytest.raises(ValueError, match="GITEA_API_TOKEN"):
+        AuthConfig()
+
+
+def test_get_auth_headers(monkeypatch):
+    """Test authentication headers generation."""
+    monkeypatch.setenv("GITEA_API_URL", "http://gitea.example.com/api/v1")
+    monkeypatch.setenv("GITEA_API_TOKEN", "test_token_123")
+
+    config = AuthConfig()
+    headers = config.get_auth_headers()
+
+    assert headers["Authorization"] == "token test_token_123"
+    assert headers["Content-Type"] == "application/json"

tests/test_client.py

@@ -0,0 +1,93 @@
+"""Tests for Gitea API client."""
+
+import pytest
+import httpx
+from unittest.mock import AsyncMock, MagicMock
+from gitea_mcp.auth import AuthConfig
+from gitea_mcp.client import (
+    GiteaClient,
+    GiteaClientError,
+    GiteaAuthError,
+    GiteaNotFoundError,
+    GiteaServerError,
+)
+
+
+@pytest.fixture
+def mock_config(monkeypatch):
+    """Create mock authentication config."""
+    monkeypatch.setenv("GITEA_API_URL", "http://gitea.example.com/api/v1")
+    monkeypatch.setenv("GITEA_API_TOKEN", "test_token_123")
+    return AuthConfig()
+
+
+@pytest.mark.asyncio
+async def test_client_initialization(mock_config):
+    """Test client initialization."""
+    client = GiteaClient(mock_config, timeout=10.0)
+    assert client.config == mock_config
+    assert client.timeout == 10.0
+
+
+@pytest.mark.asyncio
+async def test_client_context_manager(mock_config):
+    """Test client as async context manager."""
+    async with GiteaClient(mock_config) as client:
+        assert client._client is not None
+        assert isinstance(client._client, httpx.AsyncClient)
+
+
+@pytest.mark.asyncio
+async def test_client_without_context_manager_raises_error(mock_config):
+    """Test that using client without context manager raises error."""
+    client = GiteaClient(mock_config)
+
+    with pytest.raises(GiteaClientError, match="not initialized"):
+        await client.get("/test")
+
+
+@pytest.mark.asyncio
+async def test_handle_error_401(mock_config):
+    """Test handling 401 authentication error."""
+    response = MagicMock()
+    response.status_code = 401
+
+    async with GiteaClient(mock_config) as client:
+        with pytest.raises(GiteaAuthError, match="Authentication failed"):
+            client._handle_error(response)
+
+
+@pytest.mark.asyncio
+async def test_handle_error_403(mock_config):
+    """Test handling 403 forbidden error."""
+    response = MagicMock()
+    response.status_code = 403
+
+    async with GiteaClient(mock_config) as client:
+        with pytest.raises(GiteaAuthError, match="Access forbidden"):
+            client._handle_error(response)
+
+
+@pytest.mark.asyncio
+async def test_handle_error_404(mock_config):
+    """Test handling 404 not found error."""
+    response = MagicMock()
+    response.status_code = 404
+    response.request = MagicMock()
+    response.request.url = "http://gitea.example.com/api/v1/test"
+
+    async with GiteaClient(mock_config) as client:
+        with pytest.raises(GiteaNotFoundError, match="not found"):
+            client._handle_error(response)
+
+
+@pytest.mark.asyncio
+async def test_handle_error_500(mock_config):
+    """Test handling 500 server error."""
+    response = MagicMock()
+    response.status_code = 500
+    response.text = "Internal Server Error"
+
+    async with GiteaClient(mock_config) as client:
+        with pytest.raises(GiteaServerError, match="server error"):
+            client._handle_error(response)