"""Main MCP server for Google Ads integration."""
import logging
from mcp.server import Server
from mcp.server.stdio import stdio_server
from .config import settings
from .memory.database import init_database
from .knowledge.resources import register_knowledge_resources
from .actions import register_all_action_tools
from .memory import register_all_memory_tools
# Configure logging to stderr (stdout is reserved for MCP communication)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[logging.StreamHandler()],
)
logger = logging.getLogger("google-ads-mcp")
def create_server() -> Server:
"""Create and configure the MCP server.
Returns:
Configured MCP Server instance.
"""
server = Server("google-ads-mcp")
# Initialize the database for the memory layer
init_database()
logger.info("Database initialized")
# Register knowledge layer resources
register_knowledge_resources(server)
logger.info("Knowledge resources registered")
# Register action layer tools
register_all_action_tools(server)
logger.info("Action tools registered")
# Register memory layer tools
register_all_memory_tools(server)
logger.info("Memory tools registered")
# Log configuration status
if settings.is_configured:
logger.info("Google Ads credentials configured")
else:
logger.warning(
"Google Ads credentials not configured. "
"Set environment variables or create a .env file."
)
return server
async def run_server() -> None:
"""Run the MCP server with STDIO transport."""
server = create_server()
logger.info("Starting Google Ads MCP server...")
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options(),
)
def main() -> None:
"""Entry point for the MCP server."""
import asyncio
asyncio.run(run_server())
if __name__ == "__main__":
main()
