The `mge` command-line interface

The mge command-line interface is the recommended way to create, inspect, pilot and serve Endpoints.

Although every component of the mercury.graph.evidence module can be used directly from Python, most users will interact with Endpoints through the mge command. It automates the complete Endpoint lifecycle, from creating a new project to exposing it as a REST service.

Typical workflow

The most common workflow consists of the following steps:

mge new demo

# Edit the Endpoint configuration

mge pilot demo ALL_READY

mge summary demo

mge serve demo ALL_READY 8000

where:

new creates a new Endpoint project.
pilot loads the Endpoint and drives it to the desired operational state.
summary displays the current Endpoint configuration and state.
serve exposes the Endpoint through its REST API.

Command reference

The following output is produced directly by running:

mge --help

This is the authoritative description of the command-line interface. Whenever the CLI changes, this output should be considered the reference.

usage: Mercury-graph Evidence: Endpoint management cli 3.3.1 [-h] [--just_once] [--log_file LOG_FILE] [--version]
                                                                     {new,summary,pilot,serve,unlock,complete} name [intent] [port]

Creates, displays, serves and pilots persisted Endpoint objects.

positional arguments:
  {new,summary,pilot,serve,unlock,complete}
                        📁 new [name]:                 Creates the scaffold of a new Endpoint object with all the necessary files.
                        📊 summary [path]:             Displays a summary of the state of an Endpoint.
                        🌀 pilot [path, intent]:       Loads the Endpoint and pilots it to an intended state running the necessary
                                                       queries to reach that state.
                        🌎 serve [path, intent, port]: Loads the Endpoint, verifies the intent and serves it via http on the given
                                                       port. It exposes its Agentic .meta property and the .run method.
                        🔑 unlock [path]:              Forces removing the lock of the Endpoint. Use with caution!
                        ✨ complete bash:              Prints the Bash tab-completion command.
                                                       Use: source <(mge complete bash)
  name                  name of new Endpoint (for new) or path to an existing Endpoint (all other commands).
  intent                desired final state (for pilot) or required state (for serve)
  port                  port to serve the Endpoint (only for serve)

options:
  -h, --help            show this help message and exit
  --just_once           stop at first run instead of until intent is reached (only for pilot)
  --log_file LOG_FILE   path of the Agentic event log file (only for pilot and serve)
  --version             show program's version number and exit

Python implementation

`cli.mge`

`MgeCli(args)`

The MgeCli class is a command line interface for managing Mercury-graph Evidence Endpoint objects.

Parameters:

Name	Type	Description	Default
`args`	`dict`	The command line arguments as parsed by argparse.	required

Source code in cli/mge.py

def __init__(self, args):
    cmd = args['command']
    self.name = args['name']

    if cmd not in ['pilot', 'serve']:
        return

    if args['intent'] is None:
        print('Error: The intent argument is required.')
        sys.exit(1)

    self.intent = args['intent']
    self.just_once = args['just_once']
    self.logger = None

    if args.get('log_file') is not None:
        try:
            self.logger = MgeFileLogger(args['log_file'])

        except (OSError, TypeError, ValueError):
            print('Error: The log file "%s" cannot be written to.' % args['log_file'])
            sys.exit(1)

    if cmd != 'serve':
        return

    if args['port'] is None:
        print('Error: The port argument is required.')
        sys.exit(1)

    try:
        self.port = int(args['port'])

    except ValueError:
        print('Error: The port argument must be an integer.')
        sys.exit(1)

`__exec(cmd)`

Executes a command and captures the output.

Parameters:

Name	Type	Description	Default
`cmd`	`str`	The command to execute.	required

Returns:

Type	Description
`list`	The output of the command as a list of strings.

Source code in cli/mge.py

def __exec(self, cmd):
    """ Executes a command and captures the output.

    Args:
        cmd (str): The command to execute.

    Returns:
        (list): The output of the command as a list of strings.
    """

    try:
        txt = subprocess.run(cmd, shell = True, check = True, capture_output = True)

    except:
        msg = '"%s" returned an error %s' % (cmd, subprocess.CalledProcessError)
        raise RuntimeError(msg)

    return [s for s in txt.stdout.decode('utf8').strip().split('\n')]

`complete()`

Executes the "complete". The argument self.name is ignored. Should be "bash" because it is a mandatory argument.

Source code in cli/mge.py

def complete(self):
    """ Executes the "complete". The argument self.name is ignored. Should be "bash" because it is a mandatory argument. """

    print('complete -W "new summary pilot serve unlock complete ALL_READY --just_once --log_file --help --version" -A directory mge')

`new()`

Executes the "new" command after the arguments have been checked to exist.

Source code in cli/mge.py

def new(self):
    """ Executes the "new" command after the arguments have been checked to exist. """

    ofn = os.path.abspath(self.name)
    if os.path.exists(ofn):
        print('Error: The target directory "%s" already exists. Please choose a different name or edit the existing Endpoint.' % ofn)
        sys.exit(1)

    full_name = self.name.split('/')
    if len(full_name) > 1:
        pat = os.path.abspath('/'.join(full_name[:-1]))
        if not os.path.exists(pat):
            print('Error: The parent directory "%s" does not exist. Please create it first.' % pat)
            sys.exit(1)

        self.name = full_name[-1]

    if mg.evidence.Agentic._normalize_name(self.name) != self.name:
        print('Error: The name "%s" is not valid. It must be a string of letters, numbers, and underscores.' % self.name)
        sys.exit(1)

    ifn = str(pathlib.Path(__file__).resolve().parent / 'new_endpoint_template')
    if not os.path.exists(ifn):
        print('Error: The source template directory "%s" does not exist. Try re-installing the package.' % ifn)
        sys.exit(1)

    self.__exec('cp -r %s %s' % (ifn, ofn))

    conf_fn = os.path.join(ofn, 'mge_endpoint.jsonc')
    creation_date = datetime.date.today().isoformat()

    with open(conf_fn, 'r') as f:
        txt = f.read()

    txt = txt.replace('"name": ""', '"name": "%s"' % self.name, 1)
    txt = txt.replace('"creation_date": ""', '"creation_date": "%s"' % creation_date, 1)
    txt = txt.replace('"mge_version": ""', '"mge_version": "%s"' % mg.__version__, 1)

    with open(conf_fn, 'w') as f:
        f.write(txt)

    try:
        ep = mg.evidence.Endpoint(ofn)

    except Exception:
        print('Error: The newly created Endpoint "%s" failed to load.' % self.name)
        sys.exit(1)

    print ('Created a new Endpoint object "%s" in folder "%s".' % (ep.id, ep.home))

`pilot()`

Executes the "pilot" command after the arguments have been checked to exist.

Source code in cli/mge.py

def pilot(self):
    """ Executes the "pilot" command after the arguments have been checked to exist. """

    try:
        ep = mg.evidence.Endpoint(self.name, logger = self.logger)

    except Exception:
        print('Error: Could not load the Endpoint object from "%s". Please check the path and try again.' % self.name)
        sys.exit(1)

    if ep.lock(mg.evidence.endpoint.LockState.LOCK) != mg.evidence.endpoint.LockState.LOCK:
        print('Error: Could not lock the Endpoint "%s". It is locked by another process.' % ep.id)
        sys.exit(1)

    print ('Piloting the Endpoint "%s" to the state "%s" with just_once=%s ...' % (ep.id, self.intent, self.just_once))

    try:
        ep.pilot(self.intent, just_once = self.just_once)

    finally:
        ep.close(True)
        ep.lock(mg.evidence.endpoint.LockState.FREE)

    state = ep.meta['state']
    name  = ep.state_name(state)
    if name is None:
        name = '\033[2m(no name)\033[0m'

    print ('\nFinal state is %d "%s"\n\nDone.' % (state, name))

`serve()`

Executes the "serve" command after the arguments have been checked to exist.

Source code in cli/mge.py

def serve(self):
    """ Executes the "serve" command after the arguments have been checked to exist. """

    try:
        ep = mg.evidence.Endpoint(self.name, logger = self.logger)

    except Exception:
        print('Error: Could not load the Endpoint object from "%s". Please check the path and try again.' % self.name)
        sys.exit(1)

    if ep.lock(mg.evidence.endpoint.LockState.LOCK) != mg.evidence.endpoint.LockState.LOCK:
        print('Error: Could not lock the Endpoint "%s". It is locked by another process.' % ep.id)
        sys.exit(1)

    state = ep.meta['state']
    name  = ep.state_name(state)
    if name is None:
        name = '\033[2m(no name)\033[0m'

    if str(state) != str(self.intent) and str(name).lower() != str(self.intent).lower():
        ep.lock(mg.evidence.endpoint.LockState.FREE)
        print('Error: The Endpoint "%s" is in state %d "%s" but the intent is "%s".' % (ep.id, state, name, self.intent))
        sys.exit(1)

    print ('Serving the Endpoint "%s" see "http://127.0.0.1:%s/meta" ...' % (ep.id, self.port))

    try:
        MgeHttpServe(ep).serve(self.port)

    finally:
        ep.close(True)
        ep.lock(mg.evidence.endpoint.LockState.FREE)

`summary()`

Executes the "summary" command after the arguments have been checked to exist.

Source code in cli/mge.py

def summary(self):
    """ Executes the "summary" command after the arguments have been checked to exist. """

    try:
        ep = mg.evidence.Endpoint(self.name)

    except Exception:
        print('Error: Could not load the Endpoint object from "%s". Please check the path and try again.' % self.name)
        sys.exit(1)

    print (ep)

`unlock()`

Executes the "unlock" command after the arguments have been checked to exist.

Source code in cli/mge.py

def unlock(self):
    """ Executes the "unlock" command after the arguments have been checked to exist. """

    try:
        ep = mg.evidence.Endpoint(self.name, auto_pilot = False)

    except Exception:
        print('Error: Could not load the Endpoint object from "%s". Please check the path and try again.' % self.name)
        sys.exit(1)

    ep.lock(mg.evidence.endpoint.LockState.FORCE_FREE)

    print ('Endpoint "%s" forcefully unlocked.' % ep.id)

`MgeFileLogger(path)`

A minimal append-only file logger for Agentic events.

Parameters:

Name	Type	Description	Default
`path`	`str`	The path to the log file.	required

Source code in cli/mge.py

def __init__(self, path):
    self.path = path
    with open(self.path, 'a'):
        pass

`append(event)`

Appends an Agentic event dictionary to the log file.

Parameters:

Name	Type	Description	Default
`event`	`dict`	The Agentic event to log.	required

Source code in cli/mge.py

def append(self, event):
    """ Appends an Agentic event dictionary to the log file.

    Arguments:
        event (dict): The Agentic event to log.
    """

    with open(self.path, 'a') as f:
        f.write('%s\n' % event)

`MgeHttpServe(ep)`

The MgeHttpServe class exposes an Endpoint Agentic API over HTTP.

Parameters:

Name	Type	Description	Default
`ep`	`Endpoint`	The Endpoint to expose.	required

Source code in cli/mge.py

def __init__(self, ep):
    self.ep = ep
    self.app = FastAPI(title = 'Mercury-graph Evidence Endpoint', version = mg.__version__)

    self.app.get('/')(self.root)
    self.app.get('/favicon.ico', include_in_schema = False)(self.favicon)

    self.app.get('/meta')(self.meta)
    self.app.post('/run')(self.run)
    self.app.post('/dry_run')(self.dry_run)

`__validated_request(request)`

Checks that a request body is a JSON object.

Parameters:

Name	Type	Description	Default
`request`	`dict`	The JSON request body.	required

Returns:

Type	Description
`dict`	The validated request body.

Source code in cli/mge.py

def __validated_request(self, request):
    """ Checks that a request body is a JSON object.

    Args:
        request (dict): The JSON request body.

    Returns:
        (dict): The validated request body.
    """

    if type(request) != dict:
        raise HTTPException(status_code = 400, detail = 'Request body must be a JSON object.')

    return request

`dry_run(request=Body(...))`

Simulates running a request against the Endpoint.

Parameters:

Name	Type	Description	Default
`request`	`dict`	The JSON request body.	`Body(...)`

Source code in cli/mge.py

def dry_run(self, request = Body(...)):
    """ Simulates running a request against the Endpoint.

    Args:
        request (dict): The JSON request body.
    """

    request = self.__validated_request(request)

    try:
        return self.ep.dry_run(request)

    except HTTPException:
        raise

    except Exception as e:
        raise HTTPException(status_code = 500, detail = str(e))

`favicon()`

Returns the Mercury-graph favicon.

Source code in cli/mge.py

def favicon(self):
    """ Returns the Mercury-graph favicon. """

    return FileResponse('%s/favicon.ico' % pathlib.Path(__file__).resolve().parent, media_type = 'image/x-icon')

`meta()`

Returns the Endpoint metadata.

Source code in cli/mge.py

def meta(self):
    """ Returns the Endpoint metadata. """

    try:
        return self.ep.meta

    except HTTPException:
        raise

    except Exception as e:
        raise HTTPException(status_code = 500, detail = str(e))

`root()`

Redirects the root URL to the Endpoint metadata.

Source code in cli/mge.py

def root(self):
    """ Redirects the root URL to the Endpoint metadata. """

    return RedirectResponse(url = '/meta')

`run(request=Body(...))`

Runs a request against the Endpoint.

Parameters:

Name	Type	Description	Default
`request`	`dict`	The JSON request body.	`Body(...)`

Source code in cli/mge.py

def run(self, request = Body(...)):
    """ Runs a request against the Endpoint.

    Args:
        request (dict): The JSON request body.
    """

    request = self.__validated_request(request)

    try:
        return self.ep.run(request)

    except mg.evidence.agentic.AgenticRunInvalidRequest:
        raise HTTPException(status_code = 400, detail = 'Invalid request.')

    except mg.evidence.agentic.AgenticRunInvalidState:
        raise HTTPException(status_code = 503, detail = 'Invalid state.')

    except mg.evidence.agentic.AgenticRunFailed:
        raise HTTPException(status_code = 500, detail = 'Run failed.')

    except HTTPException:
        raise

    except Exception as e:
        raise HTTPException(status_code = 500, detail = str(e))

`serve(port)`

Starts the HTTP server.

Parameters:

Name	Type	Description	Default
`port`	`int`	The TCP port to listen on.	required

Source code in cli/mge.py

def serve(self, port):
    """ Starts the HTTP server.

    Args:
        port (int): The TCP port to listen on.
    """

    uvicorn.run(self.app, host = '0.0.0.0', port = port)

A minimalistic Remote Endpoint CLI example

`mercury.graph.evidence.remote.remote_endpoint`

`RemoteEndpoint(base_url)`

This is a utility class for interacting with an Endpoint that is served using the mge CLI.

It only provides a subset of the functionality and is intended for quick testing.

Parameters:

Name	Type	Description	Default
`base_url`	`str`	The URL of the endpoint as shown by the CLI (E.g., Uvicorn running on http://0.0.0.0:8765 (Press CTRL+C to quit))	required

Source code in mercury/graph/evidence/remote/remote_endpoint.py

def __init__(self, base_url):
	self.base_url = base_url

	self.capabilities = self.get_capabilities()
	self.functions	  = self.get_functions()

`dry_run(fun_name, args)`

Perform a dry run of a function on the remote endpoint.

Parameters:

Name	Type	Description	Default
`fun_name`	`str`	The name of the function to dry run.	required
`args`	`dict`	The arguments to pass to the function.	required

Returns:

Type	Description
`dict`	The result of the dry run.

Source code in mercury/graph/evidence/remote/remote_endpoint.py

def dry_run(self, fun_name, args):
	""" Perform a dry run of a function on the remote endpoint.

	Args:
		fun_name (str): The name of the function to dry run.
		args (dict): The arguments to pass to the function.

	Returns:
		(dict): The result of the dry run.
	"""

	fun = self.functions[fun_name]

	data = json.dumps({'name': fun_name, 'arguments': args}).encode('utf-8')

	req = urllib.request.Request('%s/dry_run' % self.base_url, data = data, headers = {'content-type': 'application/json'})

	with urllib.request.urlopen(req) as response:
		result = json.load(response)

	return result

`get_capabilities()`

Fetch the capabilities of the remote endpoint.

Returns:

Type	Description
`list`	A list of capabilities exposed by the remote endpoint.

Source code in mercury/graph/evidence/remote/remote_endpoint.py

def get_capabilities(self):
	""" Fetch the capabilities of the remote endpoint.

	Returns:
		(list): A list of capabilities exposed by the remote endpoint.
	"""

	with urllib.request.urlopen('%s/meta' % self.base_url) as response:
		meta = json.load(response)

	capabilities = meta.get('capabilities')

	if capabilities is None:
		raise RuntimeError('The Endpoint metadata does not expose capabilities.')

	return capabilities

`get_functions()`

Fetch the functions exposed by the remote endpoint.

Returns:

Type	Description
`dict`	A dictionary of functions with their descriptions and arguments.

Source code in mercury/graph/evidence/remote/remote_endpoint.py

def get_functions(self):
	""" Fetch the functions exposed by the remote endpoint.

	Returns:
		(dict): A dictionary of functions with their descriptions and arguments.
	"""

	functions = {}

	for cap in self.capabilities:
		assert cap['type'] == 'function'

		key = cap['function']['name']
		dsc = cap['function']['description']

		assert cap['function']['parameters']['type'] == 'object'

		req = cap['function']['parameters'].get('required', [])

		args = {}

		for nam, val in cap['function']['parameters']['properties'].items():
			args[nam] = {'typ': val['type'], 'dsc': val.get('description', ''), 'req': nam in req}

		functions[key] = {'dsc': dsc, 'args': args}

	return functions

`run(fun_name, args, easy=True)`

Run a function on the remote endpoint.

Parameters:

Name	Type	Description	Default
`fun_name`	`str`	The name of the function to run.	required
`args`	`dict`	The arguments to pass to the function.	required
`easy`	`bool`	If True, attempts to simplify argument passing and result handling. Defaults to True.	`True`

Returns:

Type	Description
`Any`	The result of the function execution, potentially simplified if `easy` is True.

Source code in mercury/graph/evidence/remote/remote_endpoint.py

def run(self, fun_name, args, easy = True):
	""" Run a function on the remote endpoint.

	Args:
		fun_name (str): The name of the function to run.
		args (dict): The arguments to pass to the function.
		easy (bool, optional): If True, attempts to simplify argument passing and result handling. Defaults to True.

	Returns:
		(Any): The result of the function execution, potentially simplified if `easy` is True.
	"""

	fun = self.functions[fun_name]

	if easy:

		if type(args) is str and len(fun['args']) == 1:
			key, val = next(iter(fun['args'].items()))
			if val['typ'] == 'string':
				args = {key: args}

	data = json.dumps({'name': fun_name, 'arguments': args}).encode('utf-8')

	req = urllib.request.Request('%s/run' % self.base_url, data = data, headers = {'content-type': 'application/json'})

	with urllib.request.urlopen(req) as response:
		result = json.load(response)

	if easy:

		if type(result) is dict and 'finish_reason' in result and 'message' in result and not 'history' in result:
			if result['finish_reason'] == 'stop':
				message = result['message']

				if type(message) is dict and 'content' in message:
					message = message['content']

				return message

	return result

The mge command-line interface

Typical workflow

Command reference

Python implementation

cli.mge

MgeCli(args)

__exec(cmd)

complete()

new()

pilot()

serve()

summary()

unlock()

MgeFileLogger(path)

append(event)

MgeHttpServe(ep)

__validated_request(request)

dry_run(request=Body(...))

favicon()

meta()

root()

run(request=Body(...))

serve(port)

A minimalistic Remote Endpoint CLI example

mercury.graph.evidence.remote.remote_endpoint

RemoteEndpoint(base_url)

dry_run(fun_name, args)

get_capabilities()

get_functions()

run(fun_name, args, easy=True)

The `mge` command-line interface

`cli.mge`

`MgeCli(args)`

`__exec(cmd)`

`complete()`

`new()`

`pilot()`

`serve()`

`summary()`

`unlock()`

`MgeFileLogger(path)`

`append(event)`

`MgeHttpServe(ep)`

`__validated_request(request)`

`dry_run(request=Body(...))`

`favicon()`

`meta()`

`root()`

`run(request=Body(...))`

`serve(port)`

`mercury.graph.evidence.remote.remote_endpoint`

`RemoteEndpoint(base_url)`

`dry_run(fun_name, args)`

`get_capabilities()`

`get_functions()`

`run(fun_name, args, easy=True)`