"""
DoWhy MCP Server CLI
Command-line interface for the DoWhy MCP server.
"""
import logging
import os
import sys
import typer
from typing import Optional
from . import __version__
from .server import create_server
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[logging.StreamHandler()]
)
logger = logging.getLogger("dowhy-mcp-server.cli")
# Create Typer app
app = typer.Typer(
name="dowhy-mcp-server",
help="DoWhy MCP Server - Professional Causal Inference Tools",
add_completion=False
)
@app.command()
def run(
transport: str = typer.Option(
"stdio",
"--transport", "-t",
help="Transport protocol (stdio, sse, streamable-http)"
),
mount_path: Optional[str] = typer.Option(
None,
"--mount-path", "-m",
help="Path to mount the server on (for HTTP transports)"
),
host: str = typer.Option(
"127.0.0.1",
"--host", "-h",
help="Host to bind to (for HTTP transports)"
),
port: int = typer.Option(
3000,
"--port", "-p",
help="Port to bind to (for HTTP transports)"
),
stateless: bool = typer.Option(
False,
"--stateless",
help="Use stateless HTTP mode (for HTTP transports)"
),
json_response: bool = typer.Option(
False,
"--json-response",
help="Use JSON response format (for HTTP transports)"
),
debug: bool = typer.Option(
False,
"--debug", "-d",
help="Enable debug logging"
)
):
"""Run the DoWhy MCP server."""
if debug:
logging.getLogger("dowhy-mcp-server").setLevel(logging.DEBUG)
logger.info(f"Starting DoWhy MCP Server v{__version__}")
logger.info(f"Transport: {transport}")
if transport in ["sse", "streamable-http"]:
logger.info(f"Host: {host}, Port: {port}")
if mount_path:
logger.info(f"Mount path: {mount_path}")
if stateless:
logger.info("Mode: Stateless HTTP")
if json_response:
logger.info("Response format: JSON")
# Create server
server = create_server(
name="DoWhy MCP Server",
version=__version__,
description="Professional Causal Inference Tools",
mount_path=mount_path,
stateless_http=stateless,
json_response=json_response
)
# Run server
try:
if transport in ["sse", "streamable-http"]:
import uvicorn
# For HTTP transports, we need to create an app and run it with uvicorn
if transport == "sse":
app = server.sse_app(mount_path)
else: # streamable-http
app = server.streamable_http_app()
uvicorn.run(app, host=host, port=port)
else:
# For stdio, just run the server
server.run(transport=transport)
except KeyboardInterrupt:
logger.info("Server stopped by user")
except Exception as e:
logger.error(f"Server error: {str(e)}")
sys.exit(1)
@app.command()
def info():
"""Display information about the DoWhy MCP server."""
server = create_server()
typer.echo(f"\nš DoWhy MCP Server v{__version__}")
typer.echo("=" * 50)
typer.echo("Professional Causal Inference Tools")
typer.echo("=" * 50)
# Tool categories
categories = {
"Causal Modeling": 6,
"Effect Estimation": 10,
"Attribution Analysis": 6,
"Root Cause Analysis": 5,
"Counterfactual Analysis": 6,
"Sensitivity Analysis": 6,
"Causal Discovery": 3
}
typer.echo("\nš Available Tool Categories:")
for category, count in categories.items():
typer.echo(f" ā¢ {category}: {count} tools")
total_tools = sum(categories.values())
typer.echo(f"\nš§° Total Tools: {total_tools}")
# Theoretical foundations
typer.echo("\nš Theoretical Foundations:")
foundations = [
"Structural Causal Models (Pearl)",
"Potential Outcomes Framework (Rubin)",
"Do-Calculus",
"Graphical Causal Models",
"Bootstrap Statistical Inference",
"Sensitivity Analysis Theory"
]
for foundation in foundations:
typer.echo(f" ā¢ {foundation}")
# Usage examples
typer.echo("\nš Quick Start:")
typer.echo(" Run server with stdio transport:")
typer.echo(" $ dowhy-mcp-server run")
typer.echo("\n Run server with HTTP transport:")
typer.echo(" $ dowhy-mcp-server run --transport streamable-http --port 3000")
typer.echo("\nš Documentation: https://dowhy-mcp.readthedocs.io/")
def main():
"""Main entry point for the CLI."""
app()
if __name__ == "__main__":
main()
