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 = '%s/new_endpoint_template' % mge_home()
    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' % mge_home(), 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)

mge_home()

Returns the path to the mge client home directory to locate its resource files.

Source code in cli/mge.py

def mge_home():
    """ Returns the path to the mge client home directory to locate its resource files. """

    path = '%s/cli/' % pathlib.Path(mg.__file__).resolve().parent.parent.parent     # mg points to <.>/mercury/graph/__init__.py
    if os.path.exists('%s/new_endpoint_template' % path):
        return path

    path = '%s/cli/' % pathlib.Path(mg.__file__).resolve().parent                   # mg points to <.>/__init__.py
    if os.path.exists('%s/new_endpoint_template' % path):
        return path

    return str(pathlib.Path(__file__).resolve().parent)

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

Attributes:

Name	Type	Description
capabilities	list	The capabilities of the remote endpoint, in the OpenAI tools format, as given by its metadata.
last_response	dict	The complete response of the last call to run(), including the history of an agent's tool calls when the Endpoint provides it. None before the first call and when the last call did not get a response.

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._parse_capabilities()
	self.last_response = None

_check_arguments(fun_name, fun, args)

Checks that the arguments of a function are a dictionary with all its required arguments and no unknown ones.

Parameters:

Name	Type	Description	Default
fun_name	str	The name of the function.	required
fun	dict	The function, as parsed by _parse_capabilities().	required
args	Any	The arguments to check.	required

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

def _check_arguments(self, fun_name, fun, args):
	""" Checks that the arguments of a function are a dictionary with all its required arguments and no unknown ones.

	Args:
		fun_name (str): The name of the function.
		fun (dict): The function, as parsed by `_parse_capabilities()`.
		args (Any): The arguments to check.
	"""

	arguments = fun['arguments']

	if len(arguments) == 0:
		hint = 'It takes no arguments: pass {}.'
	else:
		hint = 'Its arguments are: %s.' % ', '.join(arguments.keys())

	if type(args) is not dict:
		raise RemoteEndpointError('"%s" takes its arguments as a dictionary, not as %s. %s' % (fun_name, type(args).__name__, hint))

	unknown = [nam for nam in args if nam not in arguments]
	missing = [nam for nam, arg in arguments.items() if arg['required'] and nam not in args]

	problems = []

	if unknown:
		problems.append('has no argument%s %s' % ('s' if len(unknown) > 1 else '', ', '.join('"%s"' % nam for nam in unknown)))

	if missing:
		problems.append('is missing its required argument%s %s'
						% ('s' if len(missing) > 1 else '', ', '.join('"%s"' % nam for nam in missing)))

	if problems:
		raise RemoteEndpointError('"%s" %s. %s' % (fun_name, ' and '.join(problems), hint))

_error_detail(error)

Extracts the detail that the Endpoint sends with an HTTP error.

Parameters:

Name	Type	Description	Default
error	HTTPError	The error.	required

Returns:

Type	Description
str	The detail, or a sentence saying that there is none.

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

def _error_detail(self, error):
	""" Extracts the `detail` that the Endpoint sends with an HTTP error.

	Args:
		error (urllib.error.HTTPError): The error.

	Returns:
		(str): The detail, or a sentence saying that there is none.
	"""

	try:
		detail = json.loads(error.read()).get('detail')
	except Exception:
		detail = None

	if not detail:
		return 'the Endpoint gave no details.'

	return str(detail)

_function(name)

Returns a function, as parsed by _parse_capabilities(), or raises a RemoteEndpointError if it does not exist.

Parameters:

Name	Type	Description	Default
name	str	The name of the function.	required

Returns:

Type	Description
dict	The function.

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

def _function(self, name):
	""" Returns a function, as parsed by `_parse_capabilities()`, or raises a RemoteEndpointError if it does not exist.

	Args:
		name (str): The name of the function.

	Returns:
		(dict): The function.
	"""

	fun = self._functions.get(name)

	if fun is None:
		raise RemoteEndpointError('"%s" is not a capability of this Endpoint. Use list_capabilities() to see them.' % name)

	return fun

_parse_capabilities()

Parse the capabilities of the remote endpoint into a dictionary of functions, used by run() and describe().

Returns:

Type	Description
dict	A dictionary, keyed by capability name, with the description and the arguments of each capability.

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

def _parse_capabilities(self):
	""" Parse the capabilities of the remote endpoint into a dictionary of functions, used by `run()` and `describe()`.

	Returns:
		(dict): A dictionary, keyed by capability name, with the description and the arguments of each capability.
	"""

	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] = {'type': val['type'], 'description': val.get('description', ''), 'required': nam in req}

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

	return functions

_request(path, payload=None, what=None)

Sends a request to the remote endpoint and returns its decoded JSON answer, translating failures into RemoteEndpointError.

Parameters:

Name	Type	Description	Default
path	str	The path of the request, without the base URL (E.g., 'meta' or 'run').	required
payload	dict	The JSON body of a POST request. If None, the request is a GET.	None
what	str	What the request does, to start the error message (E.g., '"chat_with_reader"').	None

Returns:

Type	Description
Any	The decoded JSON answer.

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

def _request(self, path, payload = None, what = None):
	""" Sends a request to the remote endpoint and returns its decoded JSON answer, translating failures into RemoteEndpointError.

	Args:
		path (str): The path of the request, without the base URL (E.g., 'meta' or 'run').
		payload (dict, optional): The JSON body of a POST request. If None, the request is a GET.
		what (str, optional): What the request does, to start the error message (E.g., '"chat_with_reader"').

	Returns:
		(Any): The decoded JSON answer.
	"""

	url = '%s/%s' % (self.base_url, path)

	if payload is None:
		req = url
	else:
		data = json.dumps(payload).encode('utf-8')
		req	 = urllib.request.Request(url, data = data, headers = {'content-type': 'application/json'})

	try:
		with urllib.request.urlopen(req) as response:
			return json.load(response)

	except urllib.error.HTTPError as e:
		raise RemoteEndpointError('%s failed on the Endpoint (HTTP %d): %s'
								  % (what or 'The request', e.code, self._error_detail(e))) from e

	except (urllib.error.URLError, ConnectionError) as e:
		raise RemoteEndpointError('No Endpoint is answering at %s. Start it with `mge serve <endpoint> ALL_READY <port>`.'
								  % self.base_url) from e

_text_as_arguments(fun_name, fun, text)

Builds the arguments of a function from a single text, as run() does when easy is True.

Parameters:

Name	Type	Description	Default
fun_name	str	The name of the function.	required
fun	dict	The function, as parsed by _parse_capabilities().	required
text	str	The text.	required

Returns:

Type	Description
dict	The arguments, with the text as the value of the only required argument (or of the only argument, if none is required) when that argument is a string.

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

def _text_as_arguments(self, fun_name, fun, text):
	""" Builds the arguments of a function from a single text, as `run()` does when `easy` is True.

	Args:
		fun_name (str): The name of the function.
		fun (dict): The function, as parsed by `_parse_capabilities()`.
		text (str): The text.

	Returns:
		(dict): The arguments, with the text as the value of the only required argument (or of the only argument, if none is
			required) when that argument is a string.
	"""

	arguments = fun['arguments']

	names = [nam for nam, arg in arguments.items() if arg['required']]
	kind  = 'required'
	if len(names) == 0:
		names = list(arguments.keys())
		kind  = 'optional'

	if len(names) == 1 and arguments[names[0]]['type'] == 'string':
		return {names[0]: text}

	if len(names) == 0:
		raise RemoteEndpointError('"%s" takes no arguments. Pass an empty dictionary: {}.' % fun_name)

	if len(names) > 1:
		reason = 'takes %d %s arguments but only one text was given' % (len(names), kind)
	else:
		reason = 'takes its argument "%s" as %s, not as text' % (names[0], arguments[names[0]]['type'])

	raise RemoteEndpointError('"%s" %s. Pass a dictionary with its arguments: %s.' % (fun_name, reason, ', '.join(arguments.keys())))

describe(name)

Describe a capability of the remote endpoint: what it does and which arguments it takes.

Parameters:

Name	Type	Description	Default
name	str	The name of the capability, as returned by list_capabilities().	required

Returns:

Type	Description
dict	A dictionary with the keys 'name', 'description' and 'arguments'. 'arguments' maps each argument name to a dictionary with its 'type', 'description' and whether it is 'required'.

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

def describe(self, name):
	""" Describe a capability of the remote endpoint: what it does and which arguments it takes.

	Args:
		name (str): The name of the capability, as returned by `list_capabilities()`.

	Returns:
		(dict): A dictionary with the keys 'name', 'description' and 'arguments'. 'arguments' maps each argument name to a
			dictionary with its 'type', 'description' and whether it is 'required'.
	"""

	fun = self._function(name)

	arguments = {nam: dict(arg) for nam, arg in fun['arguments'].items()}

	return {'name': name, 'description': fun['description'], 'arguments': arguments}

dry_run(fun_name, args)

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

The arguments are not checked by the client: validating them is what the Endpoint does in a dry run.

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.

	The arguments are not checked by the client: validating them is what the Endpoint does in a dry run.

	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.
	"""

	self._function(fun_name)

	return self._request('dry_run', {'name': fun_name, 'arguments': args}, 'The dry run of "%s"' % fun_name)

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.
	"""

	meta = self._request('meta', what = 'Reading the metadata')

	capabilities = meta.get('capabilities')

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

	return capabilities

list_capabilities()

List the names of the capabilities exposed by the remote endpoint.

Each name is unique in the Endpoint and is the first argument of run().

Returns:

Type	Description
list	The names of the capabilities, in the order given by the Endpoint.

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

def list_capabilities(self):
	""" List the names of the capabilities exposed by the remote endpoint.

	Each name is unique in the Endpoint and is the first argument of `run()`.

	Returns:
		(list): The names of the capabilities, in the order given by the Endpoint.
	"""

	return list(self._functions.keys())

run(fun_name, args, easy=True)

Run a function on the remote endpoint.

The complete response is always stored in last_response, so the history of an agent's tool calls can be inspected after an easy call.

Parameters:

Name	Type	Description	Default
fun_name	str	The name of the function to run.	required
args	dict or str	The arguments to pass to the function. When easy is True, it can also be a string, which becomes the value of the only required argument (or of the only argument, if none is required) when that argument is a string.	required
easy	bool	If True, simplifies argument passing and returns the result itself: the message of the response, or its content for an agent (a capability in AGENT_CAPABILITIES). A response that finishes with an error raises a RemoteEndpointError instead. If False, returns the complete response. Defaults to True.	True

Returns:

Type	Description
Any	The result of the function when easy is True, the complete response otherwise.

Raises:

Type	Description
RemoteEndpointError	if the function does not exist, its arguments are not valid (checked before sending them), the Endpoint cannot be reached or answers with an HTTP error, or, when easy is True, the response finishes with an error.

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.

	The complete response is always stored in `last_response`, so the `history` of an agent's tool calls can be inspected after
	an easy call.

	Args:
		fun_name (str): The name of the function to run.
		args (dict or str): The arguments to pass to the function. When `easy` is True, it can also be a string, which becomes the
			value of the only required argument (or of the only argument, if none is required) when that argument is a string.
		easy (bool, optional): If True, simplifies argument passing and returns the result itself: the `message` of the response,
			or its `content` for an agent (a capability in `AGENT_CAPABILITIES`). A response that finishes with an error raises a
			RemoteEndpointError instead. If False, returns the complete response. Defaults to True.

	Returns:
		(Any): The result of the function when `easy` is True, the complete response otherwise.

	Raises:
		RemoteEndpointError: if the function does not exist, its arguments are not valid (checked before sending them), the
			Endpoint cannot be reached or answers with an HTTP error, or, when `easy` is True, the response finishes with an error.
	"""

	fun = self._function(fun_name)

	if easy and type(args) is str:
		args = self._text_as_arguments(fun_name, fun, args)

	self._check_arguments(fun_name, fun, args)

	self.last_response = None

	result = self._request('run', {'name': fun_name, 'arguments': args}, '"%s"' % fun_name)

	self.last_response = result

	if not easy or type(result) is not dict or 'finish_reason' not in result or 'message' not in result:
		return result

	message = result['message']

	if str(result['finish_reason']).lower().startswith('error'):
		if type(message) is dict:
			message = message.get('content', message)

		raise RemoteEndpointError('"%s" finished with an error: %s' % (fun_name, message))

	if fun_name in AGENT_CAPABILITIES and type(message) is dict:		# An agent answers with {'role': ..., 'content': ...}
		message = message.get('content')

	return message

RemoteEndpointError

Bases: Exception

Raised when a request to a remote Endpoint cannot be made or the Endpoint answers with an error.