1 Commits

Author SHA1 Message Date
1e0d896d87 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 <noreply@anthropic.com>
2026-02-03 15:05:25 -05:00
6 changed files with 583 additions and 5 deletions

View File

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

51
src/gitea_mcp/auth.py Normal file
View File

@@ -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",
}

216
src/gitea_mcp/client.py Normal file
View File

@@ -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

View File

@@ -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()

56
tests/test_auth.py Normal file
View File

@@ -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"

93
tests/test_client.py Normal file
View File

@@ -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)