oJobMCP.yaml

# Copyright 2025 Nuno Aguiar
# MCP Server Job
# This job starts a MCP server that listens for requests and executes jobs based on the requests.

include:
- oJobHTTPd.yaml

jobs:
# -------------------------
- name    : HTTP MCP Server
  typeArgs:
    noLog   : false 
    shortcut:
      name  : httpdMCP
      keyArg: port
      args  :
        description: description
        uri        : uri
        usestream  : usestream
        debug      : debug
        tplDesc    : tplDesc
        toolPrefix : toolPrefix
        fnsMeta    : fnsMeta
        fns        : fns
  help    :
    text: Starts a MCP server to handle requests with execution of jobs.
    expects: 
    - name: port 
      desc: (Number) The port to listen on.
    - name: uri
      desc: (String) The URI to handle requests. Default is "/mcp".
    - name: usestream
      desc: (Boolean) If true, returns responses as HTTP SSE stream events.
    - name: debug
      desc: (Boolean) If true, debug messages will be logged.
    - name: tplDesc
      desc: (Boolean) If true, all tool description fields in fnsMeta are evaluated as handlebars templates with the current oJob args.
    - name: toolPrefix
      desc: (String) If defined, prefixes exposed MCP tool names without changing fns/fnsMeta key mapping.
    - name: fnsMeta
      desc: (Map) Metadata for the functions available in the MCP server.
    - name: fns
      desc: (Map) Functions to be executed when called from the MCP server.
  check   :
    in: 
      description: isMap.default(__)
      port       : isNumber
      uri        : isString.default("/mcp")
      usestream  : toBoolean.isBoolean.default(false)
      debug      : toBoolean.isBoolean.default(false)
      tplDesc    : toBoolean.isBoolean.default(false)
      toolPrefix : isString.default("")
      fnsMeta    : isMap.default({})
      fns        : isMap.default({})
  deps    :
  - HTTP Start Server
  exec    : | #js
    var _applyTemplate = (meta, allStrings) => {
      if (!args.tplDesc) return meta
      var _walk = obj => {
        if (isString(obj) && allStrings) return $t(obj, args)
        if (isArray(obj)) return obj.map(_walk)
        if (isMap(obj)) {
          Object.keys(obj).forEach(k => {
            if (isString(obj[k]) && (allStrings || k == "description")) {
              obj[k] = $t(obj[k], args)
            } else {
              obj[k] = _walk(obj[k])
            }
          })
        }
        return obj
      }
      return _walk(meta)
    }

    // Ensures default server info and capabilities
    args.description = merge({
      protocolVersion: "2024-11-05",
      serverInfo: {
          name: "OpenAF",
          title: "OpenAF MCP server",
          version: "1.0.0"
      },
      capabilities: {
          prompts: {
              listChanged: true
          },
          tools: {
              listChanged: true
          }
      }
    }, args.description)
    args.description = _applyTemplate(clone(args.description), true)

    fns = {}

    var _applyTemplateToDescriptions = meta => _applyTemplate(meta, false)

    var _prefixToolName = n => (isString(args.toolPrefix) && args.toolPrefix.length > 0 ? args.toolPrefix + n : n)
    var _resolveToolName = n => {
      if (isString(args.toolPrefix) && args.toolPrefix.length > 0 && isString(n) && n.startsWith(args.toolPrefix)) {
        return n.substring(args.toolPrefix.length)
      }
      return n
    }
    var _toToolMeta = f => {
      var _meta
      if (isDef(args.fnsMeta[f])) {
        _meta = _applyTemplateToDescriptions(clone(args.fnsMeta[f]))
      } else {
        _meta = {
          name       : f,
          description: f,
          inputSchema: {}
        }
      }
      if (!isMap(_meta)) _meta = { name: f, description: String(_meta), inputSchema: {} }
      _meta.name = _prefixToolName(f)
      return _meta
    }

    if (args.usestream) {
      fns[args.uri] = req => {
        try {
          var _rpcFns = {
              initialize                 : params => {
                  if (args.debug) log(`initialize -- ${af.toCSLON(params)}`)
                  return args.description
              },
              "notifications/cancelled"  : params => ({}),
              "notifications/initialized": params => ({}),
              "ping"                     : params => {
                  if (args.debug) log(`ping -- ${af.toCSLON(params)}`)
                  return {}
              },
              "tools/call"               : params => {
                  if (args.debug) log(`tools/call -- ${af.toCSLON(params)}`)
                  var _res
                  var _toolName = _resolveToolName(params.name)
                  if (isDef(args.fns[_toolName])) {
                    try {
                      if (isString(args.fns[_toolName])) {
                        _res = $job(args.fns[_toolName], merge({ _httprequest: req }, params.arguments || params.input || {}))
                        if (isMap(_res)) {
                          delete _res.objId
                          delete _res.execid
                          // Check for _err and treat it as an error
                          if (isDef(_res._err)) {
                            var errMsg = _res._err
                            delete _res._err
                            logErr(`Error executing tool '${params.name}': ${errMsg}`)
                            _res = {
                                content: [{
                                    type: "text",
                                    text: String(errMsg)
                                }],
                                isError: true
                            }
                          }
                        }
                      } else {
                        throw "Function not found: " + _toolName
                      }
                    } catch(e) {
                      logErr(`Error executing tool '${params.name}': ${e.message}`)
                      _res = {
                          content: [{
                              type: "text",
                              text: `Error executing tool '${params.name}': ${e.message}`
                          }],
                          isError: true
                      }
                      $err(e)
                    }
                  } else {
                    _res = {
                        content: [{
                            type: "text",
                            text: `Error: Unknown tool '${params.name}'`
                        }],
                        isError: true
                    }
                  }
                  try {
                  if (isDef(_res) && _res.isError !== true && isString(_res.text)) {
                    _res = {
                        content: [{
                            type: "text",
                            text: _res.text
                        }],
                        isError: false
                    }
                  } else if (isDef(_res) && _res.isError !== true && isUnDef(_res.result)) {
                    _res = {
                        content: [{
                            type: "text",
                            text: isDef(__flags.MCPSERVER) && __flags.MCPSERVER.answerInTOON ? af.toTOON(_res) : stringify(_res, __, "")
                        }],
                        isError: false
                    }
                  } else {
                    if (isDef(_res) && _res.isError !== true) _res = _res.result
                  }
                  } catch(eee) { $err(eee) }

                  if (args.debug) log(`tools/call -- response: ${af.toCSLON(_res)}`)
                  return _res
              },
              "tools/list": params => {
                  if (args.debug) log(`tools/list -- ${af.toCSLON(params)}`)
                  var _fnsMeta = Object.keys(args.fns).map(f => _toToolMeta(f))
                  return {
                      tools: _fnsMeta
                  }
              },
              "prompts/list": params => ({})
          }

          // Process the JSON-RPC request
          var _rpcRes = ow.server.httpd.replyJSONRPC(global.__ojobHttp[args.port], req, _rpcFns, log, m => { if (args.debug) print(ansiColor("BG(230),BLACK", m)) })
          
          // Parse request to check if it's a notification
          var body = (isDef(req.files) && isDef(req.files.postData)) ? req.files.postData : __
          if (isUnDef(body) && isDef(req.data)) body = req.data
          
          var reqObj
          try {
            reqObj = jsonParse(body)
          } catch(e) {
            return { data: "Invalid JSON", mimetype: "text/plain", status: 400, header: {} }
          }

          // For notifications (no id field), return 204 No Content immediately
          if (isUnDef(reqObj.id)) {
            if (args.debug) log("Notification received: " + reqObj.method)
            return { data: "", mimetype: "text/plain", status: 204, header: {} }
          }

          // For requests with id, return SSE stream
          var pis = new java.io.PipedInputStream()
          var pos = new java.io.PipedOutputStream(pis)

          $doV(function() {
            try {
              var _payload = _rpcRes
              if (isMap(_rpcRes) && isDef(_rpcRes.data)) _payload = _rpcRes.data
              if (isString(_payload)) {
                try { _payload = jsonParse(_payload) } catch(__e) {}
              }
              
              if (isDef(_payload) && _payload !== "") {
                var msg = "event: message\n"
                msg += "data: " + (isDef(__flags.MCPSERVER) && __flags.MCPSERVER.answerInTOON ? af.toTOON(_payload) : JSON.stringify(_payload)) + "\n\n"
                pos.write(af.fromString2Bytes(msg))
                pos.flush()
              }
            } catch(e) {
              try {
                var errMsg = "event: error\n"
                errMsg += "data: " + JSON.stringify({ message: String(e) }) + "\n\n"
                pos.write(af.fromString2Bytes(errMsg))
                pos.flush()
              } catch(e2) {
                logErr("MCP stream error: " + String(e2))
              }
            } finally {
              try { pos.close() } catch(e) {}
            }
          })

          return {
            status  : 200,
            mimetype: "text/event-stream",
            stream  : pis,
            header  : {
              "Cache-Control"    : "no-cache, no-transform",
              "Connection"       : "keep-alive",
              "Content-Encoding" : "identity",
              "X-Accel-Buffering": "no"
            }
          }
        } catch(e) {
          return { data: String(e), mimetype: "text/plain", status: 500, header: {} }
        }
      }
    } else {
      fns[args.uri] = req => {
        var _rpcFns = {
            initialize                 : params => {
                if (args.debug) log(`initialize -- ${af.toCSLON(params)}`)
                return args.description
            },
            "notifications/cancelled"  : params => ({}),
            "notifications/initialized": params => ({}),
            "ping"                     : params => {
                if (args.debug) log(`ping -- ${af.toCSLON(params)}`)
                return {}
            },
            "tools/call"               : params => {
                if (args.debug) log(`tools/call -- ${af.toCSLON(params)}`)
                var _res
                var _toolName = _resolveToolName(params.name)
                if (isDef(args.fns[_toolName])) {
                  try {
                    if (isString(args.fns[_toolName])) {
                      _res = $job(args.fns[_toolName], merge({ _httprequest: req }, params.arguments || params.input || {}))
                      if (isMap(_res)) {
                        delete _res.objId
                        delete _res.execid
                        // Check for _err and treat it as an error
                        if (isDef(_res._err)) {
                          var errMsg = _res._err
                          delete _res._err
                          logErr(`Error executing tool '${params.name}': ${errMsg}`)
                          _res = {
                              content: [{
                                  type: "text",
                                  text: String(errMsg)
                              }],
                              isError: true
                          }
                        }
                      }
                    } else {
                      throw "Function not found: " + _toolName
                    }
                  } catch(e) {
                    logErr(`Error executing tool '${params.name}': ${e.message}`)
                    _res = {
                        content: [{
                            type: "text",
                            text: `Error executing tool '${params.name}': ${e.message}`
                        }],
                        isError: true
                    }
                    $err(e)
                  }
                } else {
                  _res = {
                      content: [{
                          type: "text",
                          text: `Error: Unknown tool '${params.name}'`
                      }],
                      isError: true
                  }
                }
                try {
                if (isDef(_res) && _res.isError !== true && isString(_res.text)) {
                  _res = {
                      content: [{
                          type: "text",
                          text: _res.text
                      }],
                      isError: false
                  }
                } else if (isDef(_res) && _res.isError !== true && isUnDef(_res.result)) {
                  _res = {
                      content: [{
                          type: "text",
                          text: (isDef(__flags.MCPSERVER) && __flags.MCPSERVER.answerInTOON ? af.toTOON(_res) : stringify(_res, __, ""))
                      }],
                      isError: false
                  }
                } else {
                  if (isDef(_res) && _res.isError !== true) _res = _res.result
                }
                } catch(eee) { $err(eee) }

                if (args.debug) log(`tools/call -- response: ${af.toCSLON(_res)}`)
                return _res
            },
            "tools/list": params => {
                if (args.debug) log(`tools/list -- ${af.toCSLON(params)}`)
                var _fnsMeta = Object.keys(args.fns).map(f => _toToolMeta(f))
                return {
                    tools: _fnsMeta
                }
            },
            "prompts/list": params => ({})
        }

        return ow.server.httpd.replyJSONRPC(global.__ojobHttp[args.port], req, _rpcFns, log, m => { if (args.debug) print(ansiColor("BG(230),BLACK", m)) })
      }
    }

    // Add handler for .well-known endpoint (optional OAuth discovery)
    /*fns[args.uri + "/.well-known/oauth-authorization-server"] = req => {
      return { data: "{}", mimetype: "application/json", status: 404, header: {} }
    }*/

    ow.server.httpd.route(global.__ojobHttp[args.port], merge(global.__ojobRoutes[args.port], fns))
    log("READY [" + args.port + "]")

# --------------------------
- name    : STDIO MCP Server
  typeArgs:
    noLog   : false 
    shortcut:
      name  : stdioMCP
      keyArg: description
      args  :
        debug      : debug
        tplDesc    : tplDesc
        toolPrefix : toolPrefix
        fnsMeta    : fnsMeta
        fns        : fns
  help    :
    text: Starts a MCP stdio to handle requests with execution of jobs.
    expects: 
    - name: debug
      desc: (String) If defined it will create a ndjson file with the provided name.
    - name: fnsMeta
      desc: (Map) Metadata for the functions available in the MCP server stdio.
    - name: tplDesc
      desc: (Boolean) If true, all tool description fields in fnsMeta are evaluated as handlebars templates with the current oJob args.
    - name: toolPrefix
      desc: (String) If defined, prefixes exposed MCP tool names without changing fns/fnsMeta key mapping.
    - name: fns
      desc: (Map) Functions to be executed when called from the MCP server stdio.
  check   :
    in: 
      description: isMap.default(__)
      debug      : isString.default(__)
      tplDesc    : toBoolean.isBoolean.default(false)
      toolPrefix : isString.default("")
      fnsMeta    : isMap.default({})
      fns        : isMap.default({})
  exec    : | #js
    var _applyTemplate = (meta, allStrings) => {
      if (!args.tplDesc) return meta
      var _walk = obj => {
        if (isString(obj) && allStrings) return $t(obj, args)
        if (isArray(obj)) return obj.map(_walk)
        if (isMap(obj)) {
          Object.keys(obj).forEach(k => {
            if (isString(obj[k]) && (allStrings || k == "description")) {
              obj[k] = $t(obj[k], args)
            } else {
              obj[k] = _walk(obj[k])
            }
          })
        }
        return obj
      }
      return _walk(meta)
    }

    // Ensures default server info and capabilities
    args.description = merge({
      protocolVersion: "2024-11-05",
      serverInfo: {
          name: "OpenAF",
          title: "OpenAF server",
          version: "1.0.0"
      },
      capabilities: {
          prompts: {
              listChanged: true
          },
          tools: {
              listChanged: true
          }
      }
    }, args.description)
    args.description = _applyTemplate(clone(args.description), true)

    var fs = {}
    var _applyTemplateToDescriptions = meta => _applyTemplate(meta, false)

    var _prefixToolName = n => (isString(args.toolPrefix) && args.toolPrefix.length > 0 ? args.toolPrefix + n : n)
    var _resolveToolName = n => {
      if (isString(args.toolPrefix) && args.toolPrefix.length > 0 && isString(n) && n.startsWith(args.toolPrefix)) {
        return n.substring(args.toolPrefix.length)
      }
      return n
    }
    var _toToolMeta = f => {
      var _meta
      if (isDef(args.fnsMeta[f])) {
        _meta = _applyTemplateToDescriptions(clone(args.fnsMeta[f]))
      } else {
        _meta = {
          name       : f,
          description: f,
          inputSchema: {}
        }
      }
      if (!isMap(_meta)) _meta = { name: f, description: String(_meta), inputSchema: {} }
      _meta.name = _prefixToolName(f)
      return _meta
    }

    Object.keys(args.fns).forEach(f => {
      var _toolName = _prefixToolName(f)
      fs[_toolName] = function(params) {
        var _jobName = _resolveToolName(_toolName)
        var _r = $job(args.fns[_jobName], params)
        if (isMap(_r)) {
          delete _r.objId
          delete _r.execid
          // Check for _err and treat it as an error
          if (isDef(_r._err)) {
            var errMsg = _r._err
            delete _r._err
            throw String(errMsg)
          }
        }
        return _r
      }
    })

    ow.server.mcpStdio(args.description, Object.keys(args.fns).map(f => _toToolMeta(f)), 
      fs,
      (type, msg) => {
      if (args.debug) {
        io.writeLineNDJSON(args.debug, { type: type, msg: msg })
      }
    })


# - name: ping
#   exec: | #js
#     args.text = "PONG!"

# - name: echo
#   exec: | #js
#     args.text = args.text

# - name: specialSum
#   exec: | #js
#     args.text = Number(args.a) + Number(args.b) * 2

# ojob:
#   #logToConsole: false
#   daemon      : true

# todo:
# #- (stdioMCP ): {}
# #  ((debug  )): /Applications/OpenAF/oJob-common/log.ndjson
# - (httpdStart): 17878
# - (httpdMCP  ): 17878
#   ((debug   )): true
#   ((uri     )): "/mcp"
#   ((fnsMeta)):
#     ping:
  #     name       : ping
  #     description: Pings the server
  #     inputSchema:
  #       type: object
  #       properties:
  #         text:
  #           type: string
  #           description: Text to return
  #       required: ["text"]
  #     annotations:
  #       title       : Ping
  #       readOnlyHint: false
  #       idempotentHint: true
  #   echo:
  #     name       : echo
  #     description: Echoes the input
  #     title      : echo
  #     inputSchema: 
  #       type: object
  #       properties:
  #         text:
  #           type: string
  #           description: Text to echo
  #       required: ["text"]
  #     annotations:
  #       title       : Echo
  #       readOnlyHint: false
  #       idempotentHint: true
  #   specialSum:
  #     name       : specialSum
  #     description: Performs a special sum
  #     inputSchema:
  #       type: object
  #       properties:
  #         a:
  #           type: number
  #           description: the a number of the sum
  #         b:
  #           type: number
  #           description: the b number of the sum
  #       required: ["a","b"]
  #     annotations:
  #       title: specialSum
  #       readOnlyHint: false
#   ((fns    )):
#     ping: ping
#     echo: echo
#     specialSum: specialSum