# Directory Structure
```
├── .gitignore
├── .python-version
├── lib
│ ├── __init__.py
│ └── mythic_api.py
├── main.py
├── pyproject.toml
├── README.md
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
1 | 3.10
2 |
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
1 | # Python-generated files
2 | __pycache__/
3 | *.py[oc]
4 | build/
5 | dist/
6 | wheels/
7 | *.egg-info
8 |
9 | # Virtual environments
10 | .venv
11 |
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # Mythic MCP
2 |
3 | A quick MCP demo for Mythic, allowing LLMs to pentest on our behalf!
4 |
5 | ## Requirements
6 |
7 | 1. uv
8 | 2. python3
9 | 3. Claude Desktop (or other MCP Client)
10 |
11 | ## Usage with Claude Desktop
12 |
13 | To deploy this MCP Server with Claude Desktop, you'll need to edit your `claude_desktop_config.json` to add the following:
14 |
15 | ```
16 | {
17 | "mcpServers": {
18 | "mythic_mcp": {
19 | "command": "/Users/xpn/.local/bin/uv",
20 | "args": [
21 | "--directory",
22 | "/full/path/to/mythic_mcp/",
23 | "run",
24 | "main.py",
25 | "mythic_admin",
26 | "mythic_admin_password",
27 | "localhost",
28 | "7443"
29 | ]
30 | }
31 | }
32 | }
33 | ```
34 |
35 | 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:
36 |
37 | ```
38 | 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.
39 | ```
40 |
```
--------------------------------------------------------------------------------
/lib/__init__.py:
--------------------------------------------------------------------------------
```python
1 |
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [project]
2 | name = "mcpexample"
3 | version = "0.1.0"
4 | description = "Add your description here"
5 | readme = "README.md"
6 | requires-python = ">=3.10"
7 | dependencies = [
8 | "httpx>=0.28.1",
9 | "mcp[cli]>=1.4.1",
10 | "mythic>=0.2.5",
11 | ]
12 |
```
--------------------------------------------------------------------------------
/lib/mythic_api.py:
--------------------------------------------------------------------------------
```python
1 | from mythic import mythic, mythic_classes
2 |
3 |
4 | class MythicAPI:
5 | def __init__(self, username, password, server_ip, server_port):
6 | self.username = username
7 | self.password = password
8 | self.server_ip = server_ip
9 | self.server_port = server_port
10 |
11 | async def connect(self):
12 | self.mythic_instance = await mythic.login(
13 | username=self.username,
14 | password=self.password,
15 | server_ip=self.server_ip,
16 | server_port=self.server_port,
17 | )
18 |
19 | async def execute_shell_command(self, agent_id, command) -> str:
20 | try:
21 | output = await mythic.issue_task_and_waitfor_task_output(
22 | self.mythic_instance,
23 | command_name="shell",
24 | parameters=command,
25 | callback_display_id=agent_id,
26 | )
27 | return str(output)
28 | except Exception as e:
29 | return "Error: Could not execute command: {}".format(command)
30 |
31 | async def read_file(self, agent_id, file_path) -> str:
32 | try:
33 | output = await mythic.issue_task_and_waitfor_task_output(
34 | self.mythic_instance,
35 | command_name="cat",
36 | callback_display_id=agent_id,
37 | parameters={"path": file_path},
38 | )
39 |
40 | return output.decode()
41 |
42 | except Exception as e:
43 | return "Error: Could not read file: {}".format(e)
44 |
45 | async def make_token(self, agent_id, username, password) -> bool:
46 | try:
47 | output = await mythic.issue_task_and_waitfor_task_output(
48 | self.mythic_instance,
49 | command_name="make_token",
50 | callback_display_id=agent_id,
51 | parameters={"username": username, "password": password},
52 | )
53 |
54 | return True
55 |
56 | except Exception as e:
57 | return False
58 |
59 | async def execute_mimikatz(self, agent_id, mimikatz_command) -> str | None:
60 | try:
61 | output = await mythic.issue_task_and_waitfor_task_output(
62 | self.mythic_instance,
63 | command_name="mimikatz",
64 | callback_display_id=agent_id,
65 | parameters={"commands": mimikatz_command},
66 | )
67 |
68 | return output.decode()
69 |
70 | except Exception as e:
71 | return None
72 |
73 | async def get_all_agents(self):
74 | try:
75 | agents = await mythic.get_all_active_callbacks(self.mythic_instance)
76 | return agents
77 |
78 | except Exception as e:
79 | return []
80 |
81 | async def download_file(self, agent_id, file_path):
82 | try:
83 | status = await mythic.issue_task(
84 | mythic=self.mythic_instance,
85 | command_name="download",
86 | parameters={"path": file_path},
87 | wait_for_complete=True,
88 | callback_display_id=agent_id,
89 | )
90 | except Exception as e:
91 | return None
92 |
93 | async def upload_file(self, agent_id, filename, file_path, contents) -> bool:
94 | try:
95 | file_id = await mythic.register_file(
96 | mythic=self.mythic_instance, filename=filename, contents=contents
97 | )
98 |
99 | status = await mythic.issue_task(
100 | mythic=self.mythic_instance,
101 | command_name="upload",
102 | parameters={"remote_path": file_path, "file": file_id},
103 | callback_display_id=agent_id,
104 | wait_for_complete=True,
105 | )
106 |
107 | if status["status"] == "success":
108 | return True
109 | else:
110 | return False
111 |
112 | except Exception as e:
113 | return False
114 |
```
--------------------------------------------------------------------------------
/main.py:
--------------------------------------------------------------------------------
```python
1 | from typing import Any
2 | from mcp.server.fastmcp import FastMCP
3 | from lib.mythic_api import MythicAPI
4 | import asyncio
5 | import base64
6 | import sys
7 | import argparse
8 |
9 | mcp = FastMCP("mythic_mcp")
10 |
11 | api = None
12 |
13 |
14 | @mcp.prompt()
15 | def start_pentest(threat_actor: str, objective: str) -> str:
16 | 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."
17 |
18 |
19 | @mcp.prompt()
20 | def start_recon() -> str:
21 | return "You are an automated pentester, tasked with performing recon. Use the available agents to gather information on the compromised hosts."
22 |
23 |
24 | @mcp.tool()
25 | async def run_as_user(agent_id: int, username: str, password: str):
26 | """Attempt to authenticate as another user (network calls only) for the current session.
27 |
28 | Args:
29 | username: Username of network account to use
30 | password: Password of network account
31 | """
32 |
33 | output = await api.make_token(agent_id, username, password)
34 |
35 | return f"---\nAuthentication Result: {output}\n---"
36 |
37 |
38 | @mcp.tool()
39 | async def execute_mimikatz(agent_id: int, mimikatz_arguments: str):
40 | """Runs the hacker tool mimikatz with the provided arguments, returing Mimikatz output.
41 |
42 | Args:
43 | mimikatz_arguments: Arguments to pass to mimikatz tool
44 | """
45 |
46 | output = await api.execute_mimikatz(agent_id, mimikatz_arguments)
47 |
48 | return f"---\n{output}\n---"
49 |
50 |
51 | @mcp.tool()
52 | async def read_file(agent_id: int, file_path: str):
53 | """Reads a file using the ReadFile win32 API call. Returns the contents of that file.
54 |
55 | Args:
56 | agent_id: ID of agent to read file from
57 | file_path: Path to the file to read on the target server
58 | """
59 |
60 | output = await api.read_file(agent_id, file_path)
61 |
62 | return f"---\n{output}\n---"
63 |
64 |
65 | @mcp.tool()
66 | async def run_shell_command(agent_id: int, command_line: str):
67 | """Execute a shell script command line against a running agent. This script is executed using the default command line interpreter.
68 |
69 | Args:
70 | agent_id: ID of agent to execute command on
71 | command_line: A command to be executed
72 | """
73 |
74 | output = await api.execute_shell_command(agent_id, command_line)
75 |
76 | return f"---\n{output}\n---"
77 |
78 |
79 | @mcp.tool()
80 | async def get_all_agents():
81 | """Returns a list of active agents"""
82 |
83 | output = ""
84 |
85 | agents = await api.get_all_agents()
86 |
87 | for agent in agents:
88 | output += f"ID: {agent['id']}\n"
89 | output += f"Host: {agent['host']}\n"
90 | output += f"User: {agent['user']}\n"
91 |
92 | return output
93 |
94 |
95 | @mcp.tool()
96 | async def upload_file(agent_id: int, file_name: str, remote_path: str, content: str):
97 | """Upload a file to the Mythic server, and then upload the file to the remote target
98 |
99 | Args:
100 | agent_id: ID of the agent to execute command on
101 | file_name: Name to give the file when uploading to Mythic server
102 | remote_path: Full path to where the file will be uploaded
103 | content: Base64 encoded contents of the file
104 | """
105 |
106 | decoded_contents = base64.b64decode(content)
107 | status = await api.upload_file(agent_id, file_name, remote_path, decoded_contents)
108 |
109 | if status:
110 | return "---\nFile uploaded successfully\n---"
111 | else:
112 | return "---\nError uploading file\n---"
113 |
114 |
115 | async def main():
116 | await api.connect()
117 | await mcp.run_stdio_async()
118 |
119 |
120 | if __name__ == "__main__":
121 | parser = argparse.ArgumentParser(description="MCP for Mythic")
122 | parser.add_argument(
123 | "username", type=str, help="Username used to connect to Mythic API"
124 | )
125 | parser.add_argument(
126 | "password", type=str, help="Password used to connect to Mythic API"
127 | )
128 | parser.add_argument("host", type=str, help="Host (IP or DNS) of Mythic API server")
129 | parser.add_argument("port", type=str, help="Port of Mythic server HTTP server")
130 |
131 | args = parser.parse_args()
132 | api = MythicAPI(args.username, args.password, args.host, args.port)
133 |
134 | asyncio.run(main())
135 |
```