agent0/scripts/pull-agent-identity.py

#!/usr/bin/env python3
"""
pull-agent-identity.py

Pulls agent identity from the gnommoweb Content API and writes the appropriate
persona files for the configured agent type.

Runs as an init container before the main agent process starts.
Exits 0 even on failure so the agent container still starts — the previous
identity file (if any) remains in place.

Environment variables:
  AGENT_ID          - gnommoweb agent ID (integer, required)
  AGENT_TYPE        - agent type: 'agent0' | 'hermes' (required)
  CONTENT_API_URL   - gnommoweb base URL, e.g. https://glitch.university
  CONTENT_API_KEY   - bearer token for the gnommoweb Content API
"""

import os
import sys
import json
import urllib.request
import urllib.error

AGENT_ID        = os.environ.get('AGENT_ID', '').strip()
AGENT_TYPE      = os.environ.get('AGENT_TYPE', '').strip()
CONTENT_API_URL = os.environ.get('CONTENT_API_URL', '').strip().rstrip('/')
CONTENT_API_KEY = os.environ.get('CONTENT_API_KEY', '').strip()

TAG = '[pull-agent-identity]'

# All output goes to stdout so it's always visible in docker compose logs.


def log(msg):
    print(f'{TAG} {msg}', flush=True)


def bail(msg):
    """Log and exit 0 so the main container still starts."""
    log(f'SKIP: {msg}')
    sys.exit(0)


def fail(msg):
    """Log a failure and exit 0 so the main container still starts."""
    log(f'FAILED: {msg}')
    sys.exit(0)


# ── Validate env ──────────────────────────────────────────────────────────────

if not AGENT_ID:
    bail('AGENT_ID not set')
if not AGENT_TYPE:
    bail('AGENT_TYPE not set')
if not CONTENT_API_URL:
    bail('CONTENT_API_URL not set')
if not CONTENT_API_KEY:
    bail('CONTENT_API_KEY not set')

# ── Fetch identity from gnommoweb ─────────────────────────────────────────────

url = f'{CONTENT_API_URL}/api/content/agents/{AGENT_ID}'
log(f'Fetching identity for agent {AGENT_ID} (type={AGENT_TYPE}) from {url}')

req = urllib.request.Request(
    url,
    headers={'Authorization': f'Bearer {CONTENT_API_KEY}'},
)

try:
    with urllib.request.urlopen(req, timeout=15) as resp:
        data = json.load(resp)
except urllib.error.HTTPError as e:
    fail(f'HTTP {e.code} from {url}: {e.read().decode()[:300]}')
except Exception as e:
    fail(f'Request error ({url}): {e}')

agent = data.get('agent')
if not agent:
    fail(f'Response from {url} contained no "agent" key')

log(f'Received identity — name={agent.get("name", "?")!r}')

# ── Build identity markdown ───────────────────────────────────────────────────

def build_identity_markdown(a, agent_id):
    name = a["name"]
    known_as = a.get('from_name') or name
    role = a.get('role', '')

    # Lead with an unambiguous self-identity statement so the model does not
    # default to "I am Agent Zero". The heading and opening sentence must be
    # explicit — "# Gunnar" alone reads as a subject heading about the user.
    parts = [
        f'## Your identity',
        f'Your name is {known_as}. You are not "Agent Zero" — that is the name of the '
        f'framework you run on. Your name is {known_as}.',
        f'Your numeric agent ID is {agent_id}. Use this when gutask commands ask for an agent ID.',
    ]
    if role:
        parts.append(f'Your role is: {role}.')

    if a.get('identity_document'): parts.append(f'\n## Background\n{a["identity_document"]}')
    if a.get('job_description'):   parts.append(f'\n## Job Description\n{a["job_description"]}')
    if a.get('guardrails'):        parts.append(f'\n## Guardrails\n{a["guardrails"]}')
    if a.get('best_practices'):    parts.append(f'\n## Best Practices\n{a["best_practices"]}')

    parts.append(
        f'\n## Task management\n'
        f'You have access to `gutask` for orientation, tasks, and agent letters.\n'
        f'- `gutask orient` — your orientation briefing: new letters, open tasks, session context\n'
        f'- `gutask help` — list all available commands\n'
        f'- Full runbook: `gutask skills gutask`\n'
        f'\n'
        f'`AGENT_ID` and `CONTENT_API_KEY` are already set in your environment — '
        f'gutask commands read them automatically.'
    )

    return '\n'.join(parts) + '\n'

# ── Write agent-type-specific files ──────────────────────────────────────────

if AGENT_TYPE == 'agent0':
    # Agent Zero uses profile 'agent0' which resolves prompts from usr/agents/agent0/prompts/
    # before falling back to usr/prompts/ or the built-in agents/agent0/prompts/.
    # Writing here ensures our identity overrides the default "I am Agent Zero" role.
    prompts_dir = '/a0/usr/agents/agent0/prompts'
    prompt_file = os.path.join(prompts_dir, 'agent.system.main.role.md')
    os.makedirs(prompts_dir, exist_ok=True)
    content = build_identity_markdown(agent, AGENT_ID)
    with open(prompt_file, 'w') as f:
        f.write(content)
    sections = [k for k in ('role', 'from_name', 'identity_document', 'job_description', 'guardrails', 'best_practices') if agent.get(k)]
    log(f'OK — wrote {len(content)} chars to {prompt_file} (sections: {", ".join(sections) or "name only"})')

else:
    log(f'WARNING: no handler for AGENT_TYPE={AGENT_TYPE!r} — nothing written')