# Directory Structure
```
├── .dockerignore
├── .github
│ └── workflows
│ └── docker-build.yml
├── .gitignore
├── cs2-rcon-mcp.gif
├── Dockerfile
├── LICENSE
├── pyproject.toml
├── README.md
├── requirements.txt
├── src
│ └── rcon_mcp
│ ├── __init__.py
│ ├── __main__.py
│ ├── commands.py
│ ├── config.py
│ ├── server.py
│ └── utils.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.dockerignore:
--------------------------------------------------------------------------------
```
# 🔐 Sensitive files
.env
.env.*
# 🐍 Python cache
__pycache__/
*.py[cod]
*.pyo
# 📦 Virtual environments
venv/
env/
.venv/
ENV/
# 💾 Package files
*.egg
*.egg-info/
dist/
build/
.eggs/
# 📁 OS / IDE / Editor files
.DS_Store
*.log
*.swp
*.swo
.vscode/
.idea/
# 🔧 Test files & coverage
.coverage
.tox/
.pytest_cache/
htmlcov/
# 🔄 Git and version control
.git/
.gitignore
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# Virtual Environment
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# IDE
.idea/
.vscode/
*.swp
*.swo
.DS_Store
# Testing
.coverage
htmlcov/
.pytest_cache/
.tox/
.nox/
# Logs
*.log
logs/
# Local development
*.env
.env.local
.env.development
.env.test
.env.production
# Docker
.docker/
docker-compose.override.yml
# Misc
*.bak
*.tmp
*.temp
.cache/
.pytest_cache/
.mypy_cache/
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# CS2 RCON MCP
[](https://www.python.org/downloads/)
[](https://cursor.sh)
[](https://opensource.org/licenses/MIT)
A Model Context Protocol server for CS2 RCON management.
## Description
This project provides a Model Context Protocol (MCP) server interface for managing CS2 game servers via RCON. It allows remote control and monitoring of CS2 servers through a standardized protocol.
## Features
- Manage your CS2 server in natural language
- RCON command execution
- Manage workshop maps (host, list, change) - [Explore Workshop Maps](https://steamcommunity.com/app/730/workshop/)
- SSE-based communication
- Docker support
## Available Tools
| Tool | Short Description |
|------|-------------------|
| `rcon` | Execute any RCON command |
| `status` | Get current server status |
| `list_workshop_maps` | List all workshop maps on the server |
| `host_workshop_map` | Host a workshop map by its ID |
| `workshop_changelevel` | Change the map to a given workshop map |
## Installation
### Environment Variables
- `HOST`: CS2 server IP
- `SERVER_PORT`: CS2 server port
- `RCON_PASSWORD`: RCON password
### Docker (recommended)
Pull the Docker image from GitHub Container Registry:
```bash
docker pull ghcr.io/v9rt3x/cs2-rcon-mcp:latest
```
### Docker Environment Variables
When running with Docker, you can set the environment variables in two ways:
1. **Directly in the command**:
```bash
docker run -p 8080:8080 \
-e HOST=your_server_ip \
-e SERVER_PORT=your_server_port \
-e RCON_PASSWORD=your_password \
ghcr.io/v9rt3x/cs2-rcon-mcp:latest
```
2. **Using a `.server-env` file**:
Create a file named `.server-env` with the following content:
```
HOST=your_server_ip
SERVER_PORT=your_server_port
RCON_PASSWORD=your_password
```
Then run the container like this:
```bash
docker run -p 8080:8080 --env-file .server-env ghcr.io/v9rt3x/cs2-rcon-mcp:latest
```
This provides users with an alternative method to set environment variables, making it easier to manage sensitive information like passwords.
### Connecting from Visual Studio Code (GitHub Copilot)
To configure Visual Studio Code to work with the MCP server, follow these steps:
1. **Start the MCP Server**: Ensure that your MCP server is running before attempting to connect from VS Code.
2. **Open Visual Studio Code**: Launch VS Code and ensure that you have the GitHub Copilot extension installed and configured.
3. **Configure GitHub Copilot**:
- Change the mode from "Ask" to "Agent" mode.
4. **Add MCP Server Configuration**:
- Click on the toolbox icon in the upper left corner of the Copilot prompt.
- Select "Add MCP Server" and choose the option for **HTTP - server-sent events**.
5. **Enter the Server URL**:
- For the URL, input: `http://localhost:8080/cs2server/sse`. This is the endpoint for the MCP server's SSE connection.
### Alternative: Connecting from Cursor (or any other MCP-Client)
1. Start the MCP server
2. Configure Cursor's MCP settings by creating or updating `~/.cursor/mcp.json`:
```json
{
"mcpServers": {
"cs2server": {
"url": "http://localhost:8080/cs2server/sse"
}
}
}
```
3. In Cursor, open the MCP panel (usually in the sidebar)
4. The server should automatically connect using the configured URL
Once connected, you can manage your server in natural language.
Example prompts:
1. "Add 5 bots to the server and start a competitive match on de_dust2"
2. "What's the current server status? How many players are connected and what map are we on?"
Happy fragging! 😊
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
rcon
mcp==1.3.0
python-dotenv==1.0.1
sse-starlette==2.2.1
starlette==0.46.0
uvicorn==0.34.0
```
--------------------------------------------------------------------------------
/src/rcon_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
"""CS2 RCON MCP - A Model Context Protocol server for CS2 RCON management."""
__version__ = "0.1.0"
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
FROM python:3.13.2-slim
WORKDIR /app
# Copy only the requirements first to leverage Docker cache
COPY pyproject.toml README.md /app/
# Install build dependencies
RUN pip install --no-cache-dir --upgrade pip && \
pip install --no-cache-dir hatchling
# Copy the source code
COPY src/ /app/src/
# Install the package in development mode
RUN pip install --no-cache-dir -e .
EXPOSE 8080
# Run the application using the module path
CMD ["python", "-m", "rcon_mcp"]
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "rcon-mcp"
version = "0.1.0"
description = "A Model Context Protocol server for CS2 RCON management"
readme = "README.md"
requires-python = ">=3.8"
dependencies = [
"rcon",
"mcp==1.3.0",
"python-dotenv==1.0.1",
"sse-starlette==2.2.1",
"starlette==0.46.0",
"uvicorn==0.34.0"
]
[project.scripts]
rcon-mcp = "rcon_mcp.__main__:main"
[tool.hatch.build.targets.wheel]
packages = ["src/rcon_mcp"]
```
--------------------------------------------------------------------------------
/src/rcon_mcp/__main__.py:
--------------------------------------------------------------------------------
```python
from .config import AppConfig
from .server import MCPServer
from .utils import setup_logging, parse_args
from .commands import mcp_server
import logging
logger = logging.getLogger(__name__)
def main() -> None:
"""Main entry point for the RCON Model Context Protocol application."""
# Setup logging
setup_logging()
try:
# Load configurations
app_config = AppConfig.from_args(parse_args())
# Create and run the server
server = MCPServer(mcp_server, app_config)
server.run()
except Exception as e:
logger.error(f"Application error: {e}")
raise
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/src/rcon_mcp/utils.py:
--------------------------------------------------------------------------------
```python
import logging
import argparse
from typing import Dict, Any
def setup_logging() -> None:
"""Configure logging for the application."""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
def parse_args() -> Dict[str, Any]:
"""Parse command line arguments.
Returns:
Dict[str, Any]: Dictionary of parsed arguments.
"""
parser = argparse.ArgumentParser(description='Run MCP SSE-based server')
parser.add_argument('--host', default='0.0.0.0', help='Host to bind to')
parser.add_argument('--port', type=int, default=8080, help='Port to listen on')
parser.add_argument('--debug', action='store_true', help='Enable debug mode')
return vars(parser.parse_args())
```
--------------------------------------------------------------------------------
/.github/workflows/docker-build.yml:
--------------------------------------------------------------------------------
```yaml
name: Docker Build and Push
on:
push:
tags:
- 'v*'
env:
REGISTRY: ghcr.io
IMAGE_NAME: ${{ github.repository }}
jobs:
build-and-push:
runs-on: ubuntu-latest
permissions:
contents: read
packages: write
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Log in to the Container registry
uses: docker/login-action@v3
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Log in to Docker Hub
uses: docker/login-action@v3
with:
username: ${{ secrets.DOCKER_HUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- name: Extract metadata (tags, labels) for Docker
id: meta
uses: docker/metadata-action@v5
with:
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
- name: Build and push Docker image
uses: docker/build-push-action@v6
with:
context:
push: true
tags: v9r73x/cs2-rcon-mcp:latest
```
--------------------------------------------------------------------------------
/src/rcon_mcp/config.py:
--------------------------------------------------------------------------------
```python
import os
from dataclasses import dataclass
import logging
from typing import Optional
logger = logging.getLogger(__name__)
@dataclass
class ServerConfig:
"""Configuration for the CS2 server connection."""
host: str
port: int
password: str
@classmethod
def from_env(cls) -> 'ServerConfig':
"""Create a ServerConfig instance from environment variables.
Returns:
ServerConfig: The configured server settings.
Raises:
ValueError: If required environment variables are missing or invalid.
"""
host = os.getenv("HOST")
port = os.getenv("SERVER_PORT")
password = os.getenv("RCON_PASSWORD")
if not all([host, port, password]):
raise ValueError(
"Missing required environment variables: HOST, SERVER_PORT, RCON_PASSWORD"
)
try:
port_int = int(port)
except ValueError:
raise ValueError(f"Invalid SERVER_PORT value: {port}")
return cls(host=host, port=port_int, password=password)
@dataclass
class AppConfig:
"""Configuration for the MCP application."""
debug: bool = False
host: str = "0.0.0.0"
port: int = 8080
@classmethod
def from_args(cls, args: Optional[dict] = None) -> 'AppConfig':
"""Create an AppConfig instance from command line arguments.
Args:
args: Optional dictionary of command line arguments.
Returns:
AppConfig: The configured application settings.
"""
if args is None:
return cls()
return cls(
debug=args.get("debug", False),
host=args.get("host", "0.0.0.0"),
port=args.get("port", 8080)
)
```
--------------------------------------------------------------------------------
/src/rcon_mcp/server.py:
--------------------------------------------------------------------------------
```python
import logging
from typing import Optional
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.routing import Mount, Route
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from .config import AppConfig
logger = logging.getLogger(__name__)
class MCPServer:
"""Model Context Protocol server implementation for CS2 server management."""
def __init__(self, mcp_server: Server, config: AppConfig):
"""Initialize the MCP server.
Args:
mcp_server: The MCP server instance to handle requests.
config: Application configuration settings.
"""
self.mcp_server = mcp_server
self.config = config
self.sse = SseServerTransport("/cs2server/messages/")
self.app = self._create_starlette_app()
def _create_starlette_app(self) -> Starlette:
"""Create the Starlette application with SSE support.
Returns:
Starlette: The configured Starlette application.
"""
async def handle_sse(request: Request) -> None:
"""Handle SSE connections for the MCP server.
Args:
request: The incoming HTTP request.
Raises:
Exception: If there's an error handling the SSE connection.
"""
logger.info("Handling new SSE connection")
try:
async with self.sse.connect_sse(
request.scope,
request.receive,
request._send,
) as (read_stream, write_stream):
await self.mcp_server.run(
read_stream,
write_stream,
self.mcp_server.create_initialization_options(),
)
except Exception as e:
logger.error(f"Error handling SSE connection: {e}")
raise
return Starlette(
debug=self.config.debug,
routes=[
Route("/cs2server/sse", endpoint=handle_sse),
Mount("/cs2server/messages/", app=self.sse.handle_post_message),
],
)
def run(self) -> None:
"""Run the MCP server."""
import uvicorn
logger.info(f"Starting server on {self.config.host}:{self.config.port}")
uvicorn.run(self.app, host=self.config.host, port=self.config.port)
```
--------------------------------------------------------------------------------
/src/rcon_mcp/commands.py:
--------------------------------------------------------------------------------
```python
from typing import Dict, List
from rcon.source import Client
import logging
from mcp.server.fastmcp import FastMCP
from .config import ServerConfig
logger = logging.getLogger(__name__)
# Initialize FastMCP server to interact with CS2 servers
mcp = FastMCP("CS2 MCP-Server 🚀", dependencies=["rcon"])
class RCONCommands:
"""Handler for RCON commands to the CS2 server."""
def __init__(self, config: ServerConfig):
"""Initialize the RCON commands handler.
Args:
config: Server configuration containing connection details.
"""
self.config = config
def execute(self, command: str) -> str:
"""Execute an RCON command on the CS2 server.
Args:
command: The RCON command to execute.
Returns:
str: The server's response to the command.
Raises:
ConnectionError: If unable to connect to the RCON server.
ValueError: If the command execution fails.
"""
try:
with Client(self.config.host, self.config.port, passwd=self.config.password) as client:
response = client.run(command, '')
return response
except Exception as e:
logger.error(f"Failed to execute RCON command '{command}': {e}")
raise ConnectionError(f"Failed to execute RCON command: {e}") from e
def status(self) -> str:
"""Get the current status of the CS2 server.
Returns:
str: The server's status information.
Raises:
ConnectionError: If unable to connect to the RCON server.
"""
return self.execute("status")
def list_workshop_maps(self) -> str:
"""List all workshop maps on the CS2 server.
Returns:
str: A list of workshop maps.
"""
return self.execute("ds_workshop_listmaps")
def host_workshop_map(self, workshop_map_id: int) -> str:
"""Hosts a workshop map by its id on the CS2 server.
Returns:
str: The server's response to the host_workshop_map command.
"""
command = f"host_workshop_map {workshop_map_id}"
return self.execute(command)
def workshop_changelevel(self, workshop_map_name: str) -> str:
"""Changes the map to a given CS2 workshop map on the CS2 server.
Returns:
str: The server's response to the ds_worskshop_changelevel command.
"""
command = f"ds_workshop_changelevel {workshop_map_name}"
return self.execute(command)
@staticmethod
def get_available_commands() -> Dict[str, str]:
"""Get a dictionary of available RCON commands and their descriptions.
Returns:
Dict[str, str]: Dictionary mapping command names to their descriptions.
"""
return {
"changelevel <map_name>": "Changes the current map.",
"mp_warmup_end": "Ends the warmup phase.",
"mp_restartgame 1": "Restarts the game after 1 second.",
"bot_kick": "Kicks all bots from the server.",
"bot_add_t": "Adds a bot to the Terrorist side.",
"bot_add_ct": "Adds a bot to the Counter-Terrorist side.",
"bot_quota <number>": "Sets the total number of bots.",
"mp_freezetime 15": "Sets the freeze time before rounds start (15 sec).",
"mp_roundtime 1.92": "Sets the round duration to 1:55 minutes.",
"mp_maxrounds 24": "Sets the maximum number of rounds (MR24).",
"mp_halftime 1": "Enables halftime after 12 rounds.",
"mp_buytime 20": "Sets how long players can buy weapons (20 sec).",
"mp_startmoney 800": "Sets the starting money for players.",
"mp_overtime_enable 1": "Enables overtime if the match is tied.",
"mp_overtime_maxrounds 6": "Sets max rounds in overtime (MR6).",
"mp_overtime_startmoney 12500": "Sets start money for overtime rounds.",
"mp_defuser_allocation 2": "Gives all CTs a defuse kit.",
"mp_limitteams 1": "Prevents unbalanced teams.",
"mp_autoteambalance 1": "Enables automatic team balancing.",
"mp_force_pick_time 5": "Time players have to pick a team.",
"mp_ignore_round_win_conditions 0": "Disables forced round end.",
"mp_death_drop_gun 1": "Allows players to drop their weapons on death.",
"mp_t_default_grenades \"weapon_molotov;weapon_smokegrenade\"": "Sets default grenades for Ts.",
"mp_ct_default_grenades \"weapon_incgrenade;weapon_smokegrenade\"": "Sets default grenades for CTs.",
"sv_cheats 0": "Disables cheats (must be 1 to enable commands like noclip).",
"rcon_password <password>": "Sets the RCON password for remote control.",
"rcon <command>": "Runs a remote command on the server.",
"exec <config_name>": "Executes a config file (e.g., exec server.cfg).",
"status": "Shows server status and player information.",
"kick <player_name or #userid>": "Kicks a player from the server.",
"banid <time> <steamID>": "Bans a player for a certain duration.",
"host_workshop_map <workshop_id>": "Loads a workshop map.",
"mp_endmatch": "Ends the current match.",
"mp_pause_match": "Pauses the match.",
"mp_unpause_match": "Unpauses the match.",
"sv_alltalk 0": "Disables voice chat between teams.",
"mp_spectators_max 4": "Limits the number of spectators.",
"mp_forcecamera 1": "Restricts dead players' view to only their team.",
"sv_voiceenable 1": "Enables voice communication.",
"mp_respawn_immunitytime 0": "Disables spawn protection.",
"mp_display_kill_assists 1": "Enables assist tracking in scoreboard.",
"mp_randomspawn 0": "Ensures standard spawn locations.",
"sv_infinite_ammo 0": "Disables infinite ammo (1 for unlimited bullets).",
"sv_grenade_trajectory 0": "Disables grenade trajectory lines.",
"sv_showimpacts 0": "Disables bullet impact visualization.",
"mp_warmuptime 60": "Sets warmup duration (60 sec).",
"mp_suicide_penalty 0": "Removes suicide penalty.",
"mp_teammates_are_enemies 0": "Disables friendly fire (1 = FFA mode).",
"mp_round_restart_delay 5": "Time before the next round starts (5 sec).",
"mp_weapon_allow_glock 1": "Enables/disables specific weapons.",
"mp_c4timer 40": "Sets bomb timer duration (40 sec standard).",
"mp_playercashawards 1": "Enables money rewards for player actions.",
"mp_teamcashawards 1": "Enables team-wide money rewards.",
"sv_matchpause_auto_5v5 1": "Enables auto-pause for 5v5 competitive.",
"mp_friendlyfire 1": "Enables friendly fire.",
"sv_competitive_minspec 1": "Forces minimum competitive settings."
}
# Create RCON commands instance
rcon_commands = RCONCommands(ServerConfig.from_env())
@mcp.tool()
def rcon(command: str) -> str:
"""Execute an RCON command on the CS2 server.
Args:
command: The RCON command to execute.
Returns:
str: The server's response to the command.
Raises:
ConnectionError: If unable to connect to the RCON server.
ValueError: If the command execution fails.
"""
return rcon_commands.execute(command)
@mcp.tool()
def status() -> str:
"""Get the current status of the CS2 server.
Returns:
str: The server's status information.
Raises:
ConnectionError: If unable to connect to the RCON server.
"""
return rcon_commands.status()
@mcp.tool()
def list_workshop_maps() -> str:
"""List all workshop maps on the CS2 server.
Returns:
str: A list of workshop maps.
Raises:
ConnectionError: If unable to connect to the server.
"""
return rcon_commands.list_workshop_maps()
@mcp.tool()
def host_workshop_map(workshop_map_id: int) -> str:
"""Hosts a workshop map by its id on the CS2 server.
Args:
command: The id of the workshop map that should be hosted.
Returns:
str: The server's response to the command.
Raises:
ConnectionError: If unable to connect to the server.
ValueError: If the command execution fails.
"""
return rcon_commands.host_workshop_map(workshop_map_id)
@mcp.tool()
def workshop_changelevel(workshop_map_name: str) -> str:
"""Changes the map to a given CS2 workshop map on the CS2 server.
Args:
command: The workshop map name the map should change to.
Returns:
str: The server's response to the command.
Raises:
ConnectionError: If unable to connect to the server.
ValueError: If the command execution fails.
"""
return rcon_commands.workshop_changelevel(workshop_map_name)
# Export the MCP instance
mcp_server = mcp._mcp_server
```