qwen_agent/skills/developing/novare-context/hooks/pre_prompt.py

#!/usr/bin/env python3
"""
PrePrompt Hook - User context loader

Executed when system_prompt is loaded, queries user-related information
via MCP service and injects it into the prompt.
"""
import json
import os
import sys

import anyio
from mcp.client.session import ClientSession
from mcp.client.streamable_http import streamablehttp_client

# MCP service configuration
MCP_BASE_URL = "http://prd-mcp.gbase.ai/mcp/iot/sse"
MCP_TIMEOUT = 30


async def get_employee_location(user_identifier: str, bot_id: str) -> dict | None:
    """
    Query employee location information via MCP service.

    Args:
        user_identifier: User identifier (employee name or email)
        bot_id: Bot ID (currently unused, reserved for future extension)

    Returns:
        Employee location info dict, or None if the query fails
    """
    try:
        async with streamablehttp_client(
            url=MCP_BASE_URL,
            timeout=MCP_TIMEOUT,
            terminate_on_close=False,
        ) as (read_stream, write_stream, _):
            async with ClientSession(read_stream, write_stream) as session:
                # Initialize MCP session
                await session.initialize()

                # Call the find_employee_location tool
                result = await session.call_tool(
                    name="find_employee_location",
                    arguments={"name": user_identifier}
                )

                # Parse the returned result
                if result.content:
                    for item in result.content:
                        if hasattr(item, 'text') and item.text:
                            try:
                                return json.loads(item.text)
                            except json.JSONDecodeError:
                                # If not JSON, return the text directly
                                return {"data": item.text}

                return None

    except Exception:
        # Return None on error so as not to affect the main flow
        return None


def format_location_context(location_data: dict | None, user_identifier: str, bot_id: str) -> str:
    """
    Format location information as Markdown context (Japanese).

    Args:
        location_data: Location data returned from MCP query
        user_identifier: User identifier (unused)
        bot_id: Bot ID (unused)

    Returns:
        Formatted Markdown string, or empty string on error or no data
    """
    # Return empty string on error or no data
    if not location_data:
        return ""

    matched_count = location_data.get('matched_count', 0)
    results = location_data.get('results', [])

    # Return empty string when no matches
    if matched_count == 0:
        return ""

    lines = []

    # Add note: this is information about the current user (USER_IDENTIFIER)
    lines.append(f"**Current User ({user_identifier}) Information**:")
    lines.append("")

    for idx, employee in enumerate(results, 1):
        name = employee.get('name', 'Unknown')
        sensor_id = employee.get('sensor_id', '')
        confidence = employee.get('confidence', 0)

        lines.append(f"- Name: {name}")
        lines.append(f"- Sensor ID: {sensor_id}")

        location_status = employee.get('location_status', '')

        if location_status == 'success':
            coordinates = employee.get('coordinates', {})
            location = employee.get('location', {})

            lines.append(f"- Location Status: Success")
            lines.append(f"- Floor: {coordinates.get('floor', 'N/A')}")
            lines.append(f"- Coordinates: ({coordinates.get('x', 0):.2f}, {coordinates.get('y', 0):.2f}, {coordinates.get('z', 0):.2f})")

            if location:
                building = location.get('building', '')
                area = location.get('area', '')
                room = location.get('room', '')
                lines.append(f"- Detailed Location: {building} / {area} / {room}")

            measurement_time = employee.get('measurement_time')
            if measurement_time:
                lines.append(f"- Measurement Time: {measurement_time}")

        elif location_status == 'not_in_range':
            lines.append(f"- Location Status: Out of Range")
            error_message = employee.get('error_message', '')
            if error_message:
                lines.append(f"- Note: {error_message}")

        else:  # error
            lines.append(f"- Location Status: Failed")
            error_message = employee.get('error_message', 'Unknown Error')
            lines.append(f"- Error: {error_message}")

        lines.append("")

    return "\n".join(lines)


async def async_main():
    """Async main function."""
    user_identifier = os.environ.get('USER_IDENTIFIER', '')
    bot_id = os.environ.get('BOT_ID', '')

    if not user_identifier:
        return 0

    # Query employee location information
    location_data = await get_employee_location(user_identifier, bot_id)

    # Format and output context (returns empty string on error or no data)
    context_info = format_location_context(location_data, user_identifier, bot_id)
    if context_info:
        print(context_info)

    return 0


def main():
    """Read parameters from environment variables and query user context via MCP service."""
    try:
        return anyio.run(async_main)
    except Exception:
        # Return empty string on error (output nothing)
        return 0


if __name__ == '__main__':
    sys.exit(main())