server.pyโข5.16 kB
"""Main MCP server for POEditor integration."""
import asyncio
import logging
from typing import Any, Dict, List, Optional
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.types import Tool, TextContent, INVALID_PARAMS
from mcp import McpError
from .utils.config import get_config
from .poeditor_client import POEditorClient, POEditorError
from .tools.projects import get_project_tools, PROJECT_HANDLERS
from .tools.languages import get_language_tools, LANGUAGE_HANDLERS
from .tools.terms import get_term_tools, TERM_HANDLERS
from .tools.translations import get_translation_tools, TRANSLATION_HANDLERS
from .tools.stats import get_stats_tools, STATS_HANDLERS
logger = logging.getLogger(__name__)
# Create MCP server instance
server = Server("poeditor-mcp")
# Global configuration and client
_config = None
_client = None
def get_client() -> POEditorClient:
"""Get or create POEditor client."""
global _client, _config
if _client is None:
if _config is None:
_config = get_config()
_client = POEditorClient(
api_token=_config.poeditor_api_token,
timeout=_config.request_timeout
)
return _client
@server.list_tools()
async def list_tools() -> List[Tool]:
"""List available POEditor tools."""
tools = []
# Add all tool categories
tools.extend(get_project_tools())
tools.extend(get_language_tools())
tools.extend(get_term_tools())
tools.extend(get_translation_tools())
tools.extend(get_stats_tools())
logger.info(f"Registered {len(tools)} tools")
return tools
@server.call_tool()
async def call_tool(name: str, arguments: Optional[Dict[str, Any]] = None) -> List[TextContent]:
"""Handle tool calls."""
if arguments is None:
arguments = {}
logger.info(f"Tool call: {name} with arguments: {arguments}")
# Get POEditor client
client = get_client()
# Route to appropriate handler
handler = None
# Check project handlers
if name in PROJECT_HANDLERS:
handler = PROJECT_HANDLERS[name]
# Check language handlers
elif name in LANGUAGE_HANDLERS:
handler = LANGUAGE_HANDLERS[name]
# Check term handlers
elif name in TERM_HANDLERS:
handler = TERM_HANDLERS[name]
# Check translation handlers
elif name in TRANSLATION_HANDLERS:
handler = TRANSLATION_HANDLERS[name]
# Check stats handlers
elif name in STATS_HANDLERS:
handler = STATS_HANDLERS[name]
if handler is None:
raise McpError(INVALID_PARAMS, f"Unknown tool: {name}")
try:
# Use client as context manager for proper cleanup
async with client:
result = await handler(client, arguments)
logger.info(f"Tool {name} completed successfully")
return result
except POEditorError as e:
logger.error(f"POEditor API error in {name}: {e}")
raise McpError(INVALID_PARAMS, f"POEditor API error: {e}")
except Exception as e:
logger.error(f"Unexpected error in {name}: {e}")
raise McpError(INVALID_PARAMS, f"Unexpected error: {e}")
@server.list_resources()
async def list_resources():
"""List available resources."""
# POEditor MCP doesn't use resources currently
return []
@server.read_resource()
async def read_resource(uri: str):
"""Read a resource."""
# POEditor MCP doesn't use resources currently
raise McpError(INVALID_PARAMS, f"Resource not found: {uri}")
async def main():
"""Run the MCP server."""
global _config
try:
# Load configuration
_config = get_config()
# Set up logging
logging.basicConfig(
level=getattr(logging, _config.log_level.upper()),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger.info("Starting POEditor MCP server...")
logger.info(f"Server name: {_config.server_name}")
logger.info(f"Version: {_config.version}")
# Test API token
try:
test_client = POEditorClient(_config.poeditor_api_token)
async with test_client:
projects = await test_client.list_projects()
logger.info(f"API token validated. Found {len(projects)} projects.")
except Exception as e:
logger.warning(f"API token validation failed: {e}")
logger.warning("Server will start but API calls may fail.")
# Initialize and run server
from mcp.server.stdio import stdio_server
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name=_config.server_name,
server_version=_config.version,
)
)
except Exception as e:
logger.error(f"Server error: {e}")
raise
if __name__ == "__main__":
asyncio.run(main())