From 1e0d896d87ea08c266f28b915b4981a7b3ac88a3 Mon Sep 17 00:00:00 2001 From: lmiranda Date: Tue, 3 Feb 2026 15:05:25 -0500 Subject: [PATCH] feat: implement MCP server core and authentication Implemented MCP server core infrastructure with authentication and HTTP client: - Created auth.py for API token management - Loads GITEA_API_URL and GITEA_API_TOKEN from environment - Uses python-dotenv for .env file support - Validates required configuration on initialization - Provides authentication headers for API requests - Created client.py with base HTTP client - GiteaClient class using httpx AsyncClient - Async HTTP methods: get(), post(), patch(), delete() - Comprehensive error handling for HTTP status codes - Custom exception hierarchy for different error types - Configurable timeout (default 30s) - Updated server.py with MCP server setup - Initialized MCP server with StdioServerTransport - Integrated AuthConfig and GiteaClient - Registered placeholder tool handlers (list_repositories, create_issue, create_pull_request) - Added CLI with --help and --version options - Proper error handling for configuration failures - Updated pyproject.toml - Added console script entry point: gitea-mcp - Created comprehensive unit tests - test_auth.py: Tests for AuthConfig validation and headers - test_client.py: Tests for GiteaClient initialization and error handling All acceptance criteria met: - MCP server initializes with StdioServerTransport - Authentication loads from environment variables - Base HTTP client with auth headers implemented - Error handling for API connection failures - Server reports available tools (placeholders for future issues) Closes #2 Co-Authored-By: Claude Opus 4.5 --- pyproject.toml | 3 + src/gitea_mcp/auth.py | 51 ++++++++++ src/gitea_mcp/client.py | 216 ++++++++++++++++++++++++++++++++++++++++ src/gitea_mcp/server.py | 169 ++++++++++++++++++++++++++++++- tests/test_auth.py | 56 +++++++++++ tests/test_client.py | 93 +++++++++++++++++ 6 files changed, 583 insertions(+), 5 deletions(-) create mode 100644 src/gitea_mcp/auth.py create mode 100644 src/gitea_mcp/client.py create mode 100644 tests/test_auth.py create mode 100644 tests/test_client.py diff --git a/pyproject.toml b/pyproject.toml index 8c3d1e4..599a1cb 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/gitea_mcp/auth.py b/src/gitea_mcp/auth.py new file mode 100644 index 0000000..ebffeac --- /dev/null +++ b/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", + } diff --git a/src/gitea_mcp/client.py b/src/gitea_mcp/client.py new file mode 100644 index 0000000..2c09ad8 --- /dev/null +++ b/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 diff --git a/src/gitea_mcp/server.py b/src/gitea_mcp/server.py index b9afbbb..2acae7b 100644 --- a/src/gitea_mcp/server.py +++ b/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 -# Placeholder for MCP server implementation -# TODO: Implement MCP server in future issues +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"], + }, + ), + ] + + @server.call_tool() + 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() diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..4bde5b5 --- /dev/null +++ b/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" diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..2ea1011 --- /dev/null +++ b/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)