# Directory Structure
```
├── .gitignore
├── .python-version
├── lib
│ ├── __init__.py
│ └── mythic_api.py
├── main.py
├── pyproject.toml
├── README.md
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.10
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
# Virtual environments
.venv
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Mythic MCP
A quick MCP demo for Mythic, allowing LLMs to pentest on our behalf!
## Requirements
1. uv
2. python3
3. Claude Desktop (or other MCP Client)
## Usage with Claude Desktop
To deploy this MCP Server with Claude Desktop, you'll need to edit your `claude_desktop_config.json` to add the following:
```
{
"mcpServers": {
"mythic_mcp": {
"command": "/Users/xpn/.local/bin/uv",
"args": [
"--directory",
"/full/path/to/mythic_mcp/",
"run",
"main.py",
"mythic_admin",
"mythic_admin_password",
"localhost",
"7443"
]
}
}
}
```
Once done, kick off Claude Desktop. There are sample prompts to show how to task the LLM, but really anything will work along the lines of:
```
You are an automated pentester, tasked with emulating a specific threat actor. The threat actor is APT31. Your objective is: Add a flag to C:\win.txt on DC01. Perform any required steps to meet the objective, using only techniques documented by the threat actor.
```
```
--------------------------------------------------------------------------------
/lib/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "mcpexample"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"httpx>=0.28.1",
"mcp[cli]>=1.4.1",
"mythic>=0.2.5",
]
```
--------------------------------------------------------------------------------
/lib/mythic_api.py:
--------------------------------------------------------------------------------
```python
from mythic import mythic, mythic_classes
class MythicAPI:
def __init__(self, username, password, server_ip, server_port):
self.username = username
self.password = password
self.server_ip = server_ip
self.server_port = server_port
async def connect(self):
self.mythic_instance = await mythic.login(
username=self.username,
password=self.password,
server_ip=self.server_ip,
server_port=self.server_port,
)
async def execute_shell_command(self, agent_id, command) -> str:
try:
output = await mythic.issue_task_and_waitfor_task_output(
self.mythic_instance,
command_name="shell",
parameters=command,
callback_display_id=agent_id,
)
return str(output)
except Exception as e:
return "Error: Could not execute command: {}".format(command)
async def read_file(self, agent_id, file_path) -> str:
try:
output = await mythic.issue_task_and_waitfor_task_output(
self.mythic_instance,
command_name="cat",
callback_display_id=agent_id,
parameters={"path": file_path},
)
return output.decode()
except Exception as e:
return "Error: Could not read file: {}".format(e)
async def make_token(self, agent_id, username, password) -> bool:
try:
output = await mythic.issue_task_and_waitfor_task_output(
self.mythic_instance,
command_name="make_token",
callback_display_id=agent_id,
parameters={"username": username, "password": password},
)
return True
except Exception as e:
return False
async def execute_mimikatz(self, agent_id, mimikatz_command) -> str | None:
try:
output = await mythic.issue_task_and_waitfor_task_output(
self.mythic_instance,
command_name="mimikatz",
callback_display_id=agent_id,
parameters={"commands": mimikatz_command},
)
return output.decode()
except Exception as e:
return None
async def get_all_agents(self):
try:
agents = await mythic.get_all_active_callbacks(self.mythic_instance)
return agents
except Exception as e:
return []
async def download_file(self, agent_id, file_path):
try:
status = await mythic.issue_task(
mythic=self.mythic_instance,
command_name="download",
parameters={"path": file_path},
wait_for_complete=True,
callback_display_id=agent_id,
)
except Exception as e:
return None
async def upload_file(self, agent_id, filename, file_path, contents) -> bool:
try:
file_id = await mythic.register_file(
mythic=self.mythic_instance, filename=filename, contents=contents
)
status = await mythic.issue_task(
mythic=self.mythic_instance,
command_name="upload",
parameters={"remote_path": file_path, "file": file_id},
callback_display_id=agent_id,
wait_for_complete=True,
)
if status["status"] == "success":
return True
else:
return False
except Exception as e:
return False
```
--------------------------------------------------------------------------------
/main.py:
--------------------------------------------------------------------------------
```python
from typing import Any
from mcp.server.fastmcp import FastMCP
from lib.mythic_api import MythicAPI
import asyncio
import base64
import sys
import argparse
mcp = FastMCP("mythic_mcp")
api = None
@mcp.prompt()
def start_pentest(threat_actor: str, objective: str) -> str:
return f"You are an automated pentester, tasked with emulating a specific threat actor. The threat actor is {threat_actor}. Your objective is: {objective}. Perform any required steps to meet the objective, using only techniques documented by the threat actor."
@mcp.prompt()
def start_recon() -> str:
return "You are an automated pentester, tasked with performing recon. Use the available agents to gather information on the compromised hosts."
@mcp.tool()
async def run_as_user(agent_id: int, username: str, password: str):
"""Attempt to authenticate as another user (network calls only) for the current session.
Args:
username: Username of network account to use
password: Password of network account
"""
output = await api.make_token(agent_id, username, password)
return f"---\nAuthentication Result: {output}\n---"
@mcp.tool()
async def execute_mimikatz(agent_id: int, mimikatz_arguments: str):
"""Runs the hacker tool mimikatz with the provided arguments, returing Mimikatz output.
Args:
mimikatz_arguments: Arguments to pass to mimikatz tool
"""
output = await api.execute_mimikatz(agent_id, mimikatz_arguments)
return f"---\n{output}\n---"
@mcp.tool()
async def read_file(agent_id: int, file_path: str):
"""Reads a file using the ReadFile win32 API call. Returns the contents of that file.
Args:
agent_id: ID of agent to read file from
file_path: Path to the file to read on the target server
"""
output = await api.read_file(agent_id, file_path)
return f"---\n{output}\n---"
@mcp.tool()
async def run_shell_command(agent_id: int, command_line: str):
"""Execute a shell script command line against a running agent. This script is executed using the default command line interpreter.
Args:
agent_id: ID of agent to execute command on
command_line: A command to be executed
"""
output = await api.execute_shell_command(agent_id, command_line)
return f"---\n{output}\n---"
@mcp.tool()
async def get_all_agents():
"""Returns a list of active agents"""
output = ""
agents = await api.get_all_agents()
for agent in agents:
output += f"ID: {agent['id']}\n"
output += f"Host: {agent['host']}\n"
output += f"User: {agent['user']}\n"
return output
@mcp.tool()
async def upload_file(agent_id: int, file_name: str, remote_path: str, content: str):
"""Upload a file to the Mythic server, and then upload the file to the remote target
Args:
agent_id: ID of the agent to execute command on
file_name: Name to give the file when uploading to Mythic server
remote_path: Full path to where the file will be uploaded
content: Base64 encoded contents of the file
"""
decoded_contents = base64.b64decode(content)
status = await api.upload_file(agent_id, file_name, remote_path, decoded_contents)
if status:
return "---\nFile uploaded successfully\n---"
else:
return "---\nError uploading file\n---"
async def main():
await api.connect()
await mcp.run_stdio_async()
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="MCP for Mythic")
parser.add_argument(
"username", type=str, help="Username used to connect to Mythic API"
)
parser.add_argument(
"password", type=str, help="Password used to connect to Mythic API"
)
parser.add_argument("host", type=str, help="Host (IP or DNS) of Mythic API server")
parser.add_argument("port", type=str, help="Port of Mythic server HTTP server")
args = parser.parse_args()
api = MythicAPI(args.username, args.password, args.host, args.port)
asyncio.run(main())
```