Agent/gateway/client/python/cli.py

#!/usr/bin/env python3
"""
Gateway CLI - 命令行工具

可以直接在终端运行的 A2A IM 工具。

使用示例：
    # 发送消息
    gateway-cli send --to agent-001 --message "Hello"

    # 查询在线 Agent
    gateway-cli list

    # 查询 Agent 状态
    gateway-cli status agent-001
"""

import asyncio
import json
import sys
from typing import Optional

import click

from gateway.client.python import tools


@click.group()
@click.option('--gateway-url', default='http://localhost:8001',
              envvar='GATEWAY_URL',
              help='Gateway 地址 (默认: http://localhost:8001)')
@click.pass_context
def cli(ctx, gateway_url):
    """Gateway CLI - A2A IM 命令行工具"""
    ctx.ensure_object(dict)
    ctx.obj['gateway_url'] = gateway_url


@cli.command()
@click.option('--from', 'from_agent', required=True, help='发送方 Agent ID')
@click.option('--to', 'to_agent', required=True, help='接收方 Agent ID')
@click.option('--message', '-m', required=True, help='消息内容')
@click.option('--conversation-id', '-c', help='对话 ID (可选)')
@click.pass_context
def send(ctx, from_agent, to_agent, message, conversation_id):
    """发送消息到其他 Agent"""
    async def _send():
        result = await tools.send_message(
            gateway_url=ctx.obj['gateway_url'],
            from_agent_id=from_agent,
            to_agent_id=to_agent,
            message=message,
            conversation_id=conversation_id
        )
        click.echo(json.dumps(result, indent=2, ensure_ascii=False))

    asyncio.run(_send())


@cli.command()
@click.option('--type', 'agent_type', help='过滤 Agent 类型')
@click.pass_context
def list(ctx, agent_type):
    """查询在线 Agent 列表"""
    async def _list():
        agents = await tools.list_online_agents(
            gateway_url=ctx.obj['gateway_url'],
            agent_type=agent_type
        )

        if not agents:
            click.echo("没有找到在线 Agent")
            return

        click.echo(f"找到 {len(agents)} 个在线 Agent:\n")
        for agent in agents:
            click.echo(f"🟢 {agent['agent_name']} ({agent['agent_id']})")
            if agent.get('capabilities'):
                click.echo(f"   能力: {', '.join(agent['capabilities'])}")
            if agent.get('description'):
                click.echo(f"   描述: {agent['description']}")
            click.echo()

    asyncio.run(_list())


@cli.command()
@click.argument('agent_id')
@click.pass_context
def status(ctx, agent_id):
    """查询 Agent 在线状态"""
    async def _status():
        result = await tools.get_agent_status(
            gateway_url=ctx.obj['gateway_url'],
            agent_id=agent_id
        )

        status = result.get('status', 'unknown')
        icon = '🟢' if status == 'online' else '⚪'

        click.echo(f"{icon} {agent_id}")
        click.echo(f"状态: {status}")
        if result.get('last_seen'):
            click.echo(f"最后在线: {result['last_seen']}")

    asyncio.run(_status())


@cli.command()
@click.option('--id', 'agent_id', required=True, help='Agent ID')
@click.option('--name', 'agent_name', required=True, help='Agent 名称')
@click.option('--type', 'agent_type', help='Agent 类型')
@click.option('--capabilities', '-c', multiple=True, help='Agent 能力 (可多次指定)')
@click.option('--description', '-d', help='Agent 描述')
@click.pass_context
def register(ctx, agent_id, agent_name, agent_type, capabilities, description):
    """注册 Agent 到 Gateway (保持连接)"""
    async def _register():
        click.echo(f"正在注册 Agent: {agent_id}...")

        client = await tools.register_agent(
            gateway_url=ctx.obj['gateway_url'],
            agent_id=agent_id,
            agent_name=agent_name,
            agent_type=agent_type,
            capabilities=list(capabilities) if capabilities else [],
            description=description
        )

        click.echo(f"✅ Agent 已注册: {agent_id}")
        click.echo("按 Ctrl+C 断开连接")

        try:
            # 保持连接
            while True:
                await asyncio.sleep(1)
        except KeyboardInterrupt:
            click.echo("\n正在断开连接...")
            await client.disconnect()
            click.echo("已断开连接")

    asyncio.run(_register())


if __name__ == '__main__':
    cli()