generated from personal-projects/leo-claude-mktplace
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>
This commit is contained in:
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user