Discord MCP服务器

# Start local server
python discord_server.py --transport sse --port 8000

# Server will be available at:
# http://localhost:8000/sse
# Health check: http://localhost:8000/health

# For MCP client integration
python discord_server.py --transport stdio

python discord_server.py --help

Options:
  --transport {stdio,sse}    Transport protocol (default: stdio)
  --host HOST               Host for SSE mode (default: 127.0.0.1)
  --port PORT               Port for SSE mode (default: 8000)
  --mount-path PATH         Mount path for SSE (default: /sse)
  --log-level LEVEL         Logging level: DEBUG, INFO, WARNING, ERROR (default: INFO)
  --log-format FORMAT       Log format: text or json (default: text)

Examples:
  # Run with stdio transport (for MCP clients)
  python discord_server.py --transport stdio
  
  # Run with SSE transport (local HTTP server)
  python discord_server.py --transport sse --host 0.0.0.0 --port 8000
  
  # Run with debug logging
  python discord_server.py --log-level DEBUG --transport sse

# Test with MCP dev tools
mcp dev mcp_server.py

# Run integration tests
python test_mcp_integration.py

# Test local server functionality
python test_local_server.py

{
  "mcpServers": {
    "discord-server": {
      "command": "/path/to/Discord-MCP/venv/bin/python",
      "args": ["-m", "discord_mcp.cli", "--transport", "stdio"],
      "env": {
        "DISCORD_BOT_TOKEN": "your_discord_bot_token_here",
        "DISCORD_APPLICATION_ID": "your_discord_application_id_here",
        "LOG_LEVEL": "INFO",
        "PYTHONPATH": "/path/to/Discord-MCP/src"
      }
    }
  }
}

{
  "mcpServers": {
    "discord-server": {
      "command": "python",
      "args": ["/path/to/Discord-MCP/discord_server.py", "--transport", "stdio"],
      "env": {
        "DISCORD_BOT_TOKEN": "your_discord_bot_token_here",
        "DISCORD_APPLICATION_ID": "your_discord_application_id_here",
        "LOG_LEVEL": "INFO"
      }
    }
  }
}

https://discord.com/api/oauth2/authorize?client_id=YOUR_CLIENT_ID&permissions=3072&scope=bot

https://discord.com/api/oauth2/authorize?client_id=YOUR_CLIENT_ID&permissions=1099511627776&scope=bot

# List all accessible Discord servers
guilds = await client.call_tool("list_guilds", {})

# List channels in a specific server
channels = await client.call_tool("list_channels", {
    "guild_id": "123456789012345678"
})

# Get recent messages from a channel
messages = await client.call_tool("get_messages", {
    "channel_id": "123456789012345678"
})

# Get user information
user_info = await client.call_tool("get_user_info", {
    "user_id": "987654321098765432"
})

# Through MCP client - get recent messages from a channel
messages = await client.read_resource("messages://123456789012345678")

# Get user information
user_info = await client.read_resource("user://987654321098765432")

# List servers the bot has access to
guilds = await client.read_resource("guilds://")

# Send a message to a channel
result = await client.call_tool("send_message", {
    "channel_id": "123456789012345678",
    "content": "Hello from AI assistant! 👋"
})

# Reply to a specific message
result = await client.call_tool("send_message", {
    "channel_id": "123456789012345678",
    "content": "This is a reply!",
    "reply_to_message_id": "111111111111111111"
})

# Send a direct message
result = await client.call_tool("send_dm", {
    "user_id": "123456789012345678",
    "content": "Private message from AI assistant"
})

# Read DM conversation
conversation = await client.call_tool("read_direct_messages", {
    "user_id": "123456789012345678",
    "limit": 10
})

# Delete a message (requires permissions)
result = await client.call_tool("delete_message", {
    "channel_id": "123456789012345678",
    "message_id": "111111111111111111"
})

# Edit a message (bot's own messages only)
result = await client.call_tool("edit_message", {
    "channel_id": "123456789012345678",
    "message_id": "222222222222222222",
    "new_content": "Updated message content"
})

# Timeout a user for 30 minutes
result = await client.call_tool("timeout_user", {
    "guild_id": "123456789012345678",
    "user_id": "987654321098765432",
    "duration_minutes": 30,
    "reason": "Disruptive behavior in chat"
})

# Timeout a user with default duration (10 minutes)
result = await client.call_tool("timeout_user", {
    "guild_id": "123456789012345678",
    "user_id": "987654321098765432",
    "reason": "Spam messages"
})

# Remove timeout from a user
result = await client.call_tool("untimeout_user", {
    "guild_id": "123456789012345678",
    "user_id": "987654321098765432",
    "reason": "Timeout period served"
})

# Kick a user from the server
result = await client.call_tool("kick_user", {
    "guild_id": "123456789012345678",
    "user_id": "987654321098765432",
    "reason": "Violation of server rules"
})

# Ban a user from the server
result = await client.call_tool("ban_user", {
    "guild_id": "123456789012345678",
    "user_id": "987654321098765432",
    "reason": "Repeated rule violations",
    "delete_message_days": 1  # Delete messages from last 1 day
})

# Ban a user without deleting messages
result = await client.call_tool("ban_user", {
    "guild_id": "123456789012345678",
    "user_id": "987654321098765432",
    "reason": "Serious rule violation"
})

Discord-MCP/
├── src/discord_mcp/          # Main package source
│   ├── __init__.py           # Package initialization and exports
│   ├── __main__.py           # Module entry point for `python -m discord_mcp`
│   ├── cli.py                # Command line interface with argument parsing
│   ├── config.py             # Configuration management using Pydantic Settings
│   ├── discord_client.py     # Discord API client wrapper
│   ├── resources.py          # MCP resources (read operations)
│   ├── server.py             # Main MCP server implementation with FastMCP
│   ├── tools.py              # MCP tools (write/action operations)
│   └── services/             # Service layer for business logic
│       ├── __init__.py       # Service exports (DiscordService, ContentFormatter, IDiscordService)
│       ├── interfaces.py     # Abstract service contracts (IDiscordService)
│       ├── discord_service.py # Centralized Discord operations service
│       ├── content_formatter.py # Content formatting and presentation logic
│       └── validation.py     # Validation utilities and mixins
├── tests/                    # Test suite
│   ├── services/             # Service layer tests
│   │   ├── test_discord_service.py    # DiscordService integration tests
│   │   ├── test_content_formatter.py  # ContentFormatter unit tests
│   │   ├── test_interfaces.py         # Interface compliance tests
│   │   └── test_validation.py         # Validation utility tests
│   ├── test_integration.py   # End-to-end integration tests
│   ├── test_tools.py         # MCP tools testing
│   ├── test_resources.py     # MCP resources testing
│   └── test_*.py             # Additional test modules
├── discord_server.py         # Standalone server entry point
├── mcp_server.py            # MCP dev compatible entry point
├── requirements.txt          # Dependencies
├── .env.example             # Environment template
└── README.md                # This file

class IDiscordService(ABC):
    """Abstract interface defining Discord operation contracts"""
    
    @abstractmethod
    async def get_guilds_formatted(self) -> str: ...
    
    @abstractmethod
    async def get_channels_formatted(self, guild_id: str) -> str: ...
    
    @abstractmethod
    async def get_messages_formatted(self, channel_id: str, limit: int = 50) -> str: ...
    
    # ... additional methods for user info, messaging, moderation, etc.

class DiscordService(IDiscordService):
    """Centralized Discord operations with dependency injection"""
    
    def __init__(
        self,
        discord_client: DiscordClient,
        settings: Settings,
        logger: Logger,
        content_formatter: Optional[ContentFormatter] = None,
    ):
        self._discord_client = discord_client
        self._settings = settings
        self._logger = logger
        self._content_formatter = content_formatter or ContentFormatter(settings)
    
    async def get_guilds_formatted(self) -> str:
        # Business logic: fetch and filter guilds
        guilds = await self._discord_client.get_user_guilds()
        # Presentation logic: delegate to ContentFormatter
        return self._content_formatter.format_guild_info(guilds)

class ContentFormatter:
    """Handles all Discord content formatting operations"""
    
    def __init__(self, settings: Optional[Settings] = None):
        self._settings = settings
    
    def format_guild_info(self, guilds: list) -> str:
        """Format guild information into consistent markdown structure"""
        # Centralized formatting logic for guilds
    
    def format_channel_info(self, channels: list, guild_name: str) -> str:
        """Format channel information into consistent markdown structure"""
        # Centralized formatting logic for channels
    
    # ... additional formatting methods for messages, users, timestamps, etc.

# Tools and resources use the service through dependency injection
async def list_guilds() -> str:
    ctx = server.get_context()
    discord_service: IDiscordService = ctx.request_context.lifespan_context["discord_service"]
    return await discord_service.get_guilds_formatted()

# Service unit testing
@pytest.fixture
def mock_discord_service():
    return Mock(spec=IDiscordService)

async def test_list_guilds_tool(mock_discord_service):
    # Test tools in isolation by mocking the service
    mock_discord_service.get_guilds_formatted.return_value = "# Test Guilds\n..."
    # ... test implementation

# ContentFormatter unit testing
@pytest.fixture
def content_formatter():
    return ContentFormatter()

def test_format_guild_info(content_formatter):
    guilds = [{"id": "123", "name": "Test Guild"}]
    result = content_formatter.format_guild_info(guilds)
    assert "Test Guild" in result

# Run all tests
pytest tests/

# Run with coverage
pytest tests/ --cov=src/discord_mcp

# Run integration tests
python test_mcp_integration.py

# Test local server functionality
python test_local_server.py

# Format code
black src/ tests/
isort src/ tests/

# Type checking
mypy src/

# Linting
flake8 src/ tests/

# Set debug environment
export LOG_LEVEL=DEBUG
export LOG_FORMAT=json

# Run with debug output
python discord_server.py --transport sse --log-level DEBUG 2>&1 | tee debug.log

# For SSE mode
curl http://localhost:8000/health

# Check MCP resources
python -c "
import asyncio
from mcp.client.session import ClientSession
# ... health check code
"

# Clone your fork
git clone https://github.com/your-username/Discord-MCP.git
cd Discord-MCP

# Set up development environment
python -m venv venv
source venv/bin/activate
pip install -r requirements.txt
pip install -r requirements-dev.txt  # If available

# Run tests
pytest tests/