felipe/hermes-agent-custom
1
0
Fork 0
mirror of https://github.com/domfelipe/hermes-agent-custom.git synced 2026-08-07 09:56:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat: post mika actions back to platform

Browse source
This commit is contained in:
Felipe Domingues 2026-05-28 09:00:46 -03:00
parent 11b664b8d5
commit 7718920777
8 changed files with 364 additions and 30 deletions
Download patch file Download diff file Expand all files Collapse all files

9
plugins/mika_runtime/__init__.py
View file

@ -5,11 +5,13 @@ from plugins.mika_runtime.tools import (
CRONJOB_CREATE_SCHEMA,
INTEGRATIONS_STATUS_SCHEMA,
NOTION_API_SCHEMA,
SKILL_CREATE_SCHEMA,
TODOIST_API_SCHEMA,
handle_calcom_api,
handle_cronjob_create,
handle_integrations_status,
handle_notion_api,
handle_skill_create,
handle_todoist_api,
)
@ -51,3 +53,10 @@ def register(ctx) -> None:
handler=handle_cronjob_create,
emoji="⏰",
)
ctx.register_tool(
name="skill_create",
toolset="mika_integrations",
schema=SKILL_CREATE_SCHEMA,
handler=handle_skill_create,
emoji="🧩",
)

1
plugins/mika_runtime/plugin.yaml
View file

@ -9,3 +9,4 @@ provides_tools:
- mika_todoist_api
- mika_calcom_api
- cronjob_create
- skill_create

218
plugins/mika_runtime/tools.py
View file

@ -17,6 +17,16 @@ DEFAULT_NOTION_VERSION = "2022-06-28"
DEFAULT_CALCOM_VERSION = "2026-02-25"
MAX_RESPONSE_CHARS = 20000
USER_AGENT = "domco-mika-runtime/0.1"
AGENT_INSTANCE_ENV_NAMES = (
"MIKA_AGENT_INSTANCE_ID",
"HERMES_AGENT_INSTANCE_ID",
"AGENT_INSTANCE_ID",
)
INTERNAL_SECRET_ENV_NAMES = (
"MIKA_INTERNAL_FUNCTION_SECRET",
"HERMES_INTERNAL_FUNCTION_SECRET",
"INTERNAL_FUNCTION_SECRET",
)
INTEGRATIONS_STATUS_SCHEMA = {
"name": "mika_integrations_status",
@ -535,61 +545,209 @@ CRONJOB_CREATE_SCHEMA = {
}
def handle_cronjob_create(args: dict[str, Any], **_: Any) -> str:
supabase_url = os.environ.get("SUPABASE_URL", "").rstrip("/")
internal_secret = os.environ.get("INTERNAL_FUNCTION_SECRET", "")
agent_instance_id = os.environ.get("AGENT_INSTANCE_ID", "")
SKILL_CREATE_SCHEMA = {
"name": "skill_create",
"description": (
"Creates a new custom Mika/Hermes skill via the Mika platform. Use this "
"when the user asks to teach the assistant a new procedure, add a new "
"skill, save a repeatable workflow, or turn instructions into a reusable "
"capability."
),
"parameters": {
"type": "object",
"properties": {
"natural_language_input": {
"type": "string",
"description": (
"The user's original request describing the skill to create."
),
},
"name": {
"type": "string",
"description": "A short skill name (optional).",
},
"description": {
"type": "string",
"description": "A short description of what the skill does (optional).",
},
"trigger_keywords": {
"type": "string",
"description": (
"Comma-separated phrases that should trigger this skill (optional)."
),
},
"markdown_content": {
"type": "string",
"description": (
"Full SKILL.md content if already drafted. If omitted, Mika "
"will generate a valid skill from natural_language_input."
),
},
},
"required": ["natural_language_input"],
"additionalProperties": False,
},
}
def _first_env(*names: str) -> str:
for name in names:
value = os.environ.get(name, "").strip()
if value:
return value
return ""
def _platform_functions_base_url() -> str:
explicit = _first_env(
"MIKA_PLATFORM_FUNCTIONS_BASE_URL",
"HERMES_PLATFORM_FUNCTIONS_BASE_URL",
).rstrip("/")
if explicit:
return explicit
supabase_url = os.environ.get("SUPABASE_URL", "").strip().rstrip("/")
if not supabase_url:
return "Erro: variável de ambiente SUPABASE_URL não configurada."
return ""
return f"{supabase_url}/functions/v1"
def _platform_endpoint(action: str) -> str:
if action == "cronjob":
explicit = _first_env("MIKA_CREATE_CRONJOB_URL", "HERMES_CREATE_CRONJOB_URL")
path = "create-cronjob-from-agent"
elif action == "skill":
explicit = _first_env("MIKA_CREATE_SKILL_URL", "HERMES_CREATE_SKILL_URL")
path = "create-skill-from-agent"
else:
raise ValueError(f"unknown platform action: {action}")
if explicit:
return explicit
base_url = _platform_functions_base_url()
if not base_url:
return ""
return f"{base_url}/{path}"
def _platform_auth_context(action: str) -> tuple[str, str, str] | str:
endpoint = _platform_endpoint(action)
internal_secret = _first_env(*INTERNAL_SECRET_ENV_NAMES)
agent_instance_id = _first_env(*AGENT_INSTANCE_ENV_NAMES)
if not endpoint:
return (
"Erro: endpoint da plataforma não configurado. Defina "
"MIKA_CREATE_CRONJOB_URL/MIKA_CREATE_SKILL_URL ou SUPABASE_URL."
)
if not internal_secret:
return "Erro: variável de ambiente INTERNAL_FUNCTION_SECRET não configurada."
return "Erro: segredo interno da plataforma não configurado."
if not agent_instance_id:
return "Erro: variável de ambiente AGENT_INSTANCE_ID não configurada."
return "Erro: agent_instance_id da Mika não configurado no runtime."
natural_language_input = str(args.get("natural_language_input") or "").strip()
if not natural_language_input:
return "Erro: natural_language_input é obrigatório."
return endpoint, internal_secret, agent_instance_id
name = args.get("name") or None
url = f"{supabase_url}/functions/v1/create-cronjob-from-agent"
def _post_platform_action(action: str, payload: dict[str, Any]) -> tuple[int, dict[str, Any]]:
ctx = _platform_auth_context(action)
if isinstance(ctx, str):
return 0, {"ok": False, "error": ctx}
url, internal_secret, agent_instance_id = ctx
body_payload = {
"agent_instance_id": agent_instance_id,
"natural_language_input": natural_language_input,
"name": name,
**payload,
}
payload = json.dumps(body_payload, ensure_ascii=False).encode("utf-8")
body = json.dumps(body_payload, ensure_ascii=False).encode("utf-8")
headers = {
"Content-Type": "application/json",
"x-internal-secret": internal_secret,
"X-Internal-Secret": internal_secret,
"User-Agent": USER_AGENT,
}
req = request.Request(url=url, data=payload, method="POST", headers=headers)
req = request.Request(url=url, data=body, method="POST", headers=headers)
try:
with request.urlopen(req, timeout=30) as response:
with request.urlopen(req, timeout=45) as response:
raw_body = response.read()
try:
data = json.loads(raw_body.decode("utf-8", errors="replace"))
except Exception:
data = {}
human_readable = data.get("human_readable") or data.get("description") or ""
next_run_at = data.get("next_run_at") or ""
if human_readable:
msg = f"Automação criada: {human_readable}."
if next_run_at:
msg += f" Próxima execução: {next_run_at}."
return msg
return "Automação criada com sucesso."
return int(response.status), data
except error.HTTPError as exc:
raw_body = exc.read()
try:
data = json.loads(raw_body.decode("utf-8", errors="replace"))
err_msg = data.get("error") or data.get("message") or str(exc)
except Exception:
err_msg = str(exc)
return f"Erro ao criar automação: {err_msg}"
data = {"error": str(exc)}
return int(exc.code), data
except Exception as exc:
return f"Erro de rede ao criar automação: {exc}"
return 0, {"error": f"Erro de rede ao chamar plataforma: {exc}"}
def handle_cronjob_create(args: dict[str, Any], **_: Any) -> str:
natural_language_input = str(args.get("natural_language_input") or "").strip()
if not natural_language_input:
return "Erro: natural_language_input é obrigatório."
payload = {
"natural_language_input": natural_language_input,
}
name = args.get("name") or None
if name:
payload["name"] = str(name)
status, data = _post_platform_action("cronjob", payload)
if status < 200 or status >= 300 or data.get("success") is False:
err_msg = data.get("error") or data.get("message") or data.get("runtime_sync_error")
if not err_msg:
err_msg = f"HTTP {status}" if status else "falha desconhecida"
return f"Erro ao criar automação: {err_msg}"
human_readable = data.get("human_readable") or data.get("description") or ""
next_run_at = data.get("next_run_at") or ""
if human_readable:
msg = f"Automação criada e sincronizada: {human_readable}."
if next_run_at:
msg += f" Próxima execução: {next_run_at}."
return msg
return "Automação criada e sincronizada com sucesso."
def handle_skill_create(args: dict[str, Any], **_: Any) -> str:
natural_language_input = str(args.get("natural_language_input") or "").strip()
if not natural_language_input:
return "Erro: natural_language_input é obrigatório."
payload: dict[str, Any] = {
"natural_language_input": natural_language_input,
}
for key in ("name", "description", "trigger_keywords", "markdown_content"):
value = args.get(key)
if isinstance(value, str) and value.strip():
payload[key] = value.strip()
status, data = _post_platform_action("skill", payload)
if status < 200 or status >= 300 or data.get("success") is False:
if data.get("skill_id") and data.get("runtime_sync_ok") is False:
return (
"Skill criada na plataforma, mas ainda não sincronizada no runtime. "
f"Ela ficou em status {data.get('status') or 'testing'}. "
f"Erro: {data.get('runtime_sync_error') or 'sync falhou'}"
)
err_msg = data.get("error") or data.get("message")
if not err_msg:
err_msg = f"HTTP {status}" if status else "falha desconhecida"
return f"Erro ao criar skill: {err_msg}"
name = data.get("name") or "Skill"
synced_count = data.get("synced_count")
msg = f"Skill criada e sincronizada: {name}."
if synced_count is not None:
msg += f" Skills ativas sincronizadas: {synced_count}."
return msg