Headless Everything for Personal AI：当 API 成为新的 UI

"""
Headless Content Summarizer Service
一个面向 AI Agent 的内容摘要服务
支持 API、MCP 和 CLI 三种交互方式
"""

import asyncio
import json
from typing import Optional
from dataclasses import dataclass, asdict

# ===== 1. 核心业务逻辑（headless，无 GUI） =====

@dataclass
class SummaryResult:
    """摘要结果"""
    original_length: int
    summary: str
    summary_length: int
    compression_ratio: float
    key_points: list[str]
    language: str

class HeadlessSummarizer:
    """Headless 摘要服务——只通过 API/MCP/CLI 交互"""
    
    def __init__(self, model: str = "gpt-4o-mini"):
        self.model = model
    
    async def summarize(self, text: str, max_length: int = 200, 
                        language: str = "zh") -> SummaryResult:
        """
        生成文本摘要——AI Agent 的核心调用入口
        
        Args:
            text: 原文本
            max_length: 摘要最大长度
            language: 输出语言 (zh/en)
        """
        # 实际场景中这里会调用 LLM API
        # 这里用简单算法演示 headless 服务的设计模式
        
        sentences = text.replace('。', '。<SEP>').replace('.', '.<SEP>').split('<SEP>')
        sentences = [s.strip() for s in sentences if s.strip()]
        
        # 简单提取式摘要（实际应调用 LLM）
        key_sentences = sentences[:min(3, len(sentences))]
        summary = '。'.join(key_sentences)
        
        if len(summary) > max_length:
            summary = summary[:max_length] + '...'
        
        return SummaryResult(
            original_length=len(text),
            summary=summary,
            summary_length=len(summary),
            compression_ratio=len(summary) / max(len(text), 1),
            key_points=key_sentences,
            language=language
        )

# ===== 2. REST API 接口（AI Agent 调用方式一） =====

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Headless Summarizer API")
summarizer = HeadlessSummarizer()

class SummarizeRequest(BaseModel):
    text: str
    max_length: int = 200
    language: str = "zh"

@app.post("/summarize", response_model=dict)
async def api_summarize(req: SummarizeRequest):
    """AI Agent 通过 REST API 调用摘要服务"""
    result = await summarizer.summarize(
        text=req.text,
        max_length=req.max_length,
        language=req.language
    )
    return asdict(result)

@app.get("/health")
async def health_check():
    """健康检查端点"""
    return {"status": "ok", "service": "headless-summarizer"}

# ===== 3. MCP Server 接口（AI Agent 调用方式二） =====

# pip install mcp
async def run_mcp_server():
    """将服务暴露为 MCP Server，AI Agent 可直接发现和调用"""
    from mcp.server.fastmcp import FastMCP
    
    mcp = FastMCP("headless-summarizer")
    
    @mcp.tool()
    async def summarize_text(text: str, max_length: int = 200) -> dict:
        """为 AI Agent 提供文本摘要能力
        
        Args:
            text: 需要摘要的文本
            max_length: 摘要最大字符数
        """
        result = await summarizer.summarize(text, max_length)
        return asdict(result)
    
    @mcp.tool()
    async def extract_key_points(text: str, num_points: int = 5) -> list[str]:
        """从文本中提取关键要点
        
        Args:
            text: 源文本
            num_points: 提取的要点数量
        """
        result = await summarizer.summarize(text)
        return result.key_points[:num_points]
    
    await mcp.run()

# ===== 4. CLI 接口（AI Agent 调用方式三） =====

import click

@click.group()
def cli():
    """Headless Summarizer CLI - AI Agent 通过命令行调用"""
    pass

@cli.command()
@click.argument('text')
@click.option('--max-length', default=200, help='摘要最大长度')
@click.option('--language', default='zh', help='输出语言')
def summarize(text: str, max_length: int, language: str):
    """通过 CLI 生成文本摘要"""
    result = asyncio.run(summarizer.summarize(text, max_length, language))
    click.echo(json.dumps(asdict(result), ensure_ascii=False, indent=2))

@cli.command()
@click.argument('file', type=click.File('r'))
def batch_summarize(file):
    """批量处理文件中的文本"""
    text = file.read()
    result = asyncio.run(summarizer.summarize(text))
    click.echo(f"原文: {result.original_length} 字")
    click.echo(f"摘要: {result.summary}")

if __name__ == "__main__":
    # 三种启动方式：
    # python headless_summarizer.py api      -> 启动 REST API
    # python headless_summarizer.py mcp      -> 启动 MCP Server
    # python headless_summarizer.py summarize "文本" -> CLI 模式
    import sys
    if len(sys.argv) > 1:
        if sys.argv[1] == "api":
            import uvicorn
            uvicorn.run(app, host="0.0.0.0", port=8000)
        elif sys.argv[1] == "mcp":
            asyncio.run(run_mcp_server())
        else:
            cli()
    else:
        print("Usage: python headless_summarizer.py [api|mcp|<command>]")

"""
AI Agent 消费 Headless 服务的完整示例
演示 Agent 如何组合多个 headless 服务完成复杂任务
"""

import asyncio
import httpx

class AIAgent:
    """模拟 AI Agent 消费多个 headless 服务"""
    
    def __init__(self):
        self.services = {
            "summarizer": "http://localhost:8000",
            "weather": "https://api.weather.example.com",
            "calendar": "https://api.calendar.example.com",
        }
    
    async def daily_briefing(self) -> str:
        """
        AI Agent 自主组合多个 headless 服务，
        生成每日简报——这是 headless 服务的典型使用场景
        """
        # Step 1: 调用天气服务获取今日天气
        weather = await self._get_weather("Beijing")
        
        # Step 2: 调用日历服务获取今日日程
        schedule = await self._get_today_schedule()
        
        # Step 3: 调用摘要服务压缩长文本信息
        news_digest = await self._summarize_news()
        
        # Step 4: AI Agent 综合所有信息生成简报
        briefing = f"""📋 今日简报

🌤️ 天气：{weather['condition']}，{weather['temp']}°C
📅 日程：今日有 {len(schedule)} 个会议
📰 新闻摘要：
{news_digest['summary']}

💡 建议：{'今天适合户外活动' if weather['temp'] > 15 else '建议室内办公'}
"""
        return briefing
    
    async def _get_weather(self, city: str) -> dict:
        """调用 headless 天气服务"""
        # 实际场景中通过 API 调用
        return {"condition": "晴", "temp": 22, "city": city}
    
    async def _get_today_schedule(self) -> list[dict]:
        """调用 headless 日历服务"""
        return [
            {"time": "09:00", "title": "产品评审会"},
            {"time": "14:00", "title": "代码审查"},
            {"time": "16:00", "title": "技术分享"},
        ]
    
    async def _summarize_news(self) -> dict:
        """调用 headless 摘要服务"""
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.services['summarizer']}/summarize",
                json={
                    "text": "AI Agent 市场规模预计 2026 年达到 500 亿美元...",
                    "max_length": 100
                }
            )
            return resp.json()

# ===== 运行示例 =====
async def main():
    agent = AIAgent()
    briefing = await agent.daily_briefing()
    print(briefing)

if __name__ == "__main__":
    asyncio.run(main())