generated from personal-projects/leo-claude-mktplace
Compare commits
1 Commits
feat/1-pro
...
feat/2-mcp
| Author | SHA1 | Date | |
|---|---|---|---|
| 1e0d896d87 |
@@ -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"
|
||||
|
||||
51
src/gitea_mcp/auth.py
Normal file
51
src/gitea_mcp/auth.py
Normal 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
216
src/gitea_mcp/client.py
Normal 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
|
||||
@@ -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
|
||||
# TODO: Implement MCP server in future issues
|
||||
@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()
|
||||
|
||||
56
tests/test_auth.py
Normal file
56
tests/test_auth.py
Normal 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
93
tests/test_client.py
Normal 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)
|
||||
Reference in New Issue
Block a user