Hiddenden/AegisGitea-MCP
3
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
Latte
2026-02-11 18:16:00 +01:00
parent dd7bbd1f9a
commit d82fe87113
25 changed files with 120 additions and 4230 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/aegis_gitea_mcp/audit.py
View File

@@ -25,6 +25,7 @@ class AuditLogger:
# Ensure log directory exists
self.log_path.parent.mkdir(parents=True, exist_ok=True)
self._log_file = self._get_log_file()
# Configure structlog for audit logging
structlog.configure(
@@ -35,7 +36,7 @@ class AuditLogger:
],
wrapper_class=structlog.BoundLogger,
context_class=dict,
logger_factory=structlog.PrintLoggerFactory(file=self._get_log_file()),
logger_factory=structlog.PrintLoggerFactory(file=self._log_file),
cache_logger_on_first_use=True,
)
@@ -45,6 +46,13 @@ class AuditLogger:
"""Get file handle for audit log."""
return open(self.log_path, "a", encoding="utf-8")
def close(self) -> None:
"""Close open audit log resources."""
try:
self._log_file.close()
except Exception:
pass
def log_tool_invocation(
self,
tool_name: str,
@@ -166,4 +174,6 @@ def get_audit_logger() -> AuditLogger:
def reset_audit_logger() -> None:
"""Reset global audit logger instance (primarily for testing)."""
global _audit_logger
if _audit_logger is not None:
_audit_logger.close()
_audit_logger = None

18
src/aegis_gitea_mcp/config.py
View File

@@ -1,7 +1,7 @@
"""Configuration management for AegisGitea MCP server."""
from pathlib import Path
from typing import List, Optional
from typing import Optional
from pydantic import Field, HttpUrl, field_validator, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -112,20 +112,20 @@ class Settings(BaseSettings):
def validate_and_parse_api_keys(self) -> "Settings":
"""Parse and validate API keys if authentication is enabled."""
# Parse comma-separated keys into list
keys = []
keys: list[str] = []
if self.mcp_api_keys_raw and self.mcp_api_keys_raw.strip():
keys = [key.strip() for key in self.mcp_api_keys_raw.split(",") if key.strip()]
# Store in a property we'll access
object.__setattr__(self, "_mcp_api_keys", keys)
# Validate if auth is enabled
if self.auth_enabled and not keys:
raise ValueError(
"At least one API key must be configured when auth_enabled=True. "
"Set MCP_API_KEYS environment variable or disable auth with AUTH_ENABLED=false"
)
# Validate key format (at least 32 characters for security)
for key in keys:
if len(key) < 32:
@@ -133,11 +133,11 @@ class Settings(BaseSettings):
f"API keys must be at least 32 characters long. "
f"Use scripts/generate_api_key.py to generate secure keys."
)
return self
@property
def mcp_api_keys(self) -> List[str]:
def mcp_api_keys(self) -> list[str]:
"""Get parsed list of API keys."""
return getattr(self, "_mcp_api_keys", [])
@@ -155,7 +155,7 @@ def get_settings() -> Settings:
"""Get or create global settings instance."""
global _settings
if _settings is None:
_settings = Settings() # type: ignore
_settings = Settings()
return _settings

24
src/aegis_gitea_mcp/gitea_client.py
View File

@@ -69,7 +69,7 @@ class GiteaClient:
if self.client:
await self.client.aclose()
def _handle_response(self, response: Response, correlation_id: str) -> Dict[str, Any]:
def _handle_response(self, response: Response, correlation_id: str) -> Any:
"""Handle Gitea API response and raise appropriate exceptions.
Args:
@@ -148,12 +148,12 @@ class GiteaClient:
return user_data
except Exception as e:
except Exception as exc:
self.audit.log_tool_invocation(
tool_name="get_current_user",
correlation_id=correlation_id,
result_status="error",
error=str(e),
error=str(exc),
)
raise
@@ -177,7 +177,7 @@ class GiteaClient:
try:
response = await self.client.get("/api/v1/user/repos")
repos_data = self._handle_response(response, correlation_id)
# Ensure we have a list
repos = repos_data if isinstance(repos_data, list) else []
@@ -190,12 +190,12 @@ class GiteaClient:
return repos
except Exception as e:
except Exception as exc:
self.audit.log_tool_invocation(
tool_name="list_repositories",
correlation_id=correlation_id,
result_status="error",
error=str(e),
error=str(exc),
)
raise
@@ -236,13 +236,13 @@ class GiteaClient:
return repo_data
except Exception as e:
except Exception as exc:
self.audit.log_tool_invocation(
tool_name="get_repository",
repository=repo_id,
correlation_id=correlation_id,
result_status="error",
error=str(e),
error=str(exc),
)
raise
@@ -314,14 +314,14 @@ class GiteaClient:
return file_data
except Exception as e:
except Exception as exc:
self.audit.log_tool_invocation(
tool_name="get_file_contents",
repository=repo_id,
target=filepath,
correlation_id=correlation_id,
result_status="error",
error=str(e),
error=str(exc),
)
raise
@@ -370,12 +370,12 @@ class GiteaClient:
return tree_data
except Exception as e:
except Exception as exc:
self.audit.log_tool_invocation(
tool_name="get_tree",
repository=repo_id,
correlation_id=correlation_id,
result_status="error",
error=str(e),
error=str(exc),
)
raise

9
src/aegis_gitea_mcp/mcp_protocol.py
View File

@@ -2,7 +2,7 @@
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
from pydantic import BaseModel, ConfigDict, Field
class MCPTool(BaseModel):
@@ -14,9 +14,10 @@ class MCPTool(BaseModel):
..., alias="inputSchema", description="JSON Schema for tool input"
)
class Config:
populate_by_name = True
by_alias = True
model_config = ConfigDict(
populate_by_name=True,
serialize_by_alias=True,
)
class MCPToolCallRequest(BaseModel):

7
src/aegis_gitea_mcp/server.py
View File

@@ -40,9 +40,7 @@ app = FastAPI(
)
# Global settings and audit logger
# Note: auth_validator is fetched dynamically in middleware to support test resets
settings = get_settings()
audit = get_audit_logger()
# Note: access settings/audit logger dynamically to support test resets.
# Tool dispatcher mapping
@@ -115,6 +113,7 @@ async def authenticate_request(request: Request, call_next):
@app.on_event("startup")
async def startup_event() -> None:
"""Initialize server on startup."""
settings = get_settings()
logger.info(f"Starting AegisGitea MCP Server on {settings.mcp_host}:{settings.mcp_port}")
logger.info(f"Connected to Gitea instance: {settings.gitea_base_url}")
logger.info(f"Audit logging enabled: {settings.audit_log_path}")
@@ -180,6 +179,7 @@ async def call_tool(request: MCPToolCallRequest) -> JSONResponse:
Returns:
JSON response with tool execution result
"""
audit = get_audit_logger()
correlation_id = request.correlation_id or audit.log_tool_invocation(
tool_name=request.tool,
params=request.arguments,
@@ -312,6 +312,7 @@ async def sse_message_handler(request: Request) -> JSONResponse:
JSON response acknowledging the message
"""
try:
audit = get_audit_logger()
body = await request.json()
logger.info(f"Received MCP message via SSE POST: {body}")

16
src/aegis_gitea_mcp/tools/repository.py
View File

@@ -40,8 +40,8 @@ async def list_repositories_tool(gitea: GiteaClient, arguments: Dict[str, Any])
"count": len(simplified_repos),
}
except GiteaError as e:
raise Exception(f"Failed to list repositories: {str(e)}")
except GiteaError as exc:
raise RuntimeError(f"Failed to list repositories: {exc}") from exc
async def get_repository_info_tool(
@@ -84,8 +84,8 @@ async def get_repository_info_tool(
"clone_url": repo_data.get("clone_url", ""),
}
except GiteaError as e:
raise Exception(f"Failed to get repository info: {str(e)}")
except GiteaError as exc:
raise RuntimeError(f"Failed to get repository info: {exc}") from exc
async def get_file_tree_tool(gitea: GiteaClient, arguments: Dict[str, Any]) -> Dict[str, Any]:
@@ -129,8 +129,8 @@ async def get_file_tree_tool(gitea: GiteaClient, arguments: Dict[str, Any]) -> D
"count": len(simplified_tree),
}
except GiteaError as e:
raise Exception(f"Failed to get file tree: {str(e)}")
except GiteaError as exc:
raise RuntimeError(f"Failed to get file tree: {exc}") from exc
async def get_file_contents_tool(gitea: GiteaClient, arguments: Dict[str, Any]) -> Dict[str, Any]:
@@ -185,5 +185,5 @@ async def get_file_contents_tool(gitea: GiteaClient, arguments: Dict[str, Any])
"url": file_data.get("html_url", ""),
}
except GiteaError as e:
raise Exception(f"Failed to get file contents: {str(e)}")
except GiteaError as exc:
raise RuntimeError(f"Failed to get file contents: {exc}") from exc