Headless AI 全景：从 API-first 到无头服务的 AI 代理革命

# headless_crm_server.py
# 一个支持 MCP 协议的 Headless CRM 服务
from mcp.server.fastmcp import FastMCP
from dataclasses import dataclass
from typing import Optional
import json

# 创建 MCP 服务器
crm = FastMCP("HeadlessCRM")

@dataclass
class Lead:
    id: int
    name: str
    email: str
    company: str
    status: str = "new"
    score: int = 0

# 内存数据库（生产环境替换为 PostgreSQL）
leads_db: dict[int, Lead] = {}
next_id = 1

@crm.tool()
def create_lead(name: str, email: str, company: str) -> str:
    """创建新的销售线索
    
    Args:
        name: 联系人姓名
        email: 邮箱地址
        company: 公司名称
    
    Returns:
        JSON 格式的创建结果，包含 lead_id
    """
    global next_id
    lead = Lead(id=next_id, name=name, email=email, company=company)
    leads_db[next_id] = lead
    next_id += 1
    return json.dumps({
        "lead_id": lead.id,
        "status": "created",
        "lead": {"name": lead.name, "company": lead.company}
    })

@crm.tool()
def score_lead(lead_id: int) -> str:
    """AI 自动评分：根据线索信息计算意向分数
    
    评分规则：
    - 企业邮箱 (+20 分)
    - 知名公司域名 (+15 分)
    - 邮箱长度合理 (+5 分)
    """
    lead = leads_db.get(lead_id)
    if not lead:
        return json.dumps({"error": "Lead not found"})
    
    score = 50  # 基础分
    
    # 企业邮箱检测
    if not lead.email.endswith(("@gmail.com", "@yahoo.com", "@hotmail.com")):
        score += 20
    
    # 知名公司域名
    enterprise_domains = ["apple.com", "google.com", "microsoft.com"]
    domain = lead.email.split("@")[-1]
    if domain in enterprise_domains:
        score += 15
    
    # 邮箱长度合理性
    if 10 <= len(lead.email) <= 40:
        score += 5
    
    lead.score = min(score, 100)
    leads_db[lead_id] = lead
    
    return json.dumps({
        "lead_id": lead_id,
        "score": lead.score,
        "grade": "A" if score >= 80 else "B" if score >= 60 else "C"
    })

@crm.tool()
def get_leads(status: Optional[str] = None) -> str:
    """获取所有线索，可选按状态筛选"""
    filtered = list(leads_db.values())
    if status:
        filtered = [l for l in filtered if l.status == status]
    return json.dumps([
        {"id": l.id, "name": l.name, "company": l.company, "score": l.score}
        for l in filtered
    ])

if __name__ == "__main__":
    # 以 stdio 方式运行 MCP 服务器
    crm.run(transport="stdio")

# ai_agent_client.py
# AI 代理通过 MCP 协议调用 Headless CRM 服务
import subprocess
import json

class HeadlessCRMClient:
    """AI 代理的 CRM 客户端，通过 stdio 连接 MCP 服务"""
    
    def __init__(self, server_script: str):
        self.process = subprocess.Popen(
            ["python", server_script],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        )
    
    def call_tool(self, tool_name: str, **params) -> dict:
        """调用 MCP 工具"""
        request = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": tool_name,
                "arguments": params
            }
        }
        
        self.process.stdin.write(json.dumps(request) + "\n")
        self.process.stdin.flush()
        
        response = json.loads(self.process.stdout.readline())
        return response["result"]
    
    def discover_tools(self) -> list:
        """发现可用工具"""
        request = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/list"
        }
        self.process.stdin.write(json.dumps(request) + "\n")
        self.process.stdin.flush()
        response = json.loads(self.process.stdout.readline())
        return response["result"]["tools"]
    
    def close(self):
        self.process.terminate()

# === AI 代理的自主工作流 ===
if __name__ == "__main__":
    client = HeadlessCRMClient("headless_crm_server.py")
    
    # 1. 发现可用工具
    tools = client.discover_tools()
    print(f"🔍 发现 {len(tools)} 个可用工具:")
    for t in tools:
        print(f"  - {t['name']}: {t['description'][:50]}...")
    
    # 2. AI 代理自主创建线索
    new_leads = [
        {"name": "张明", "email": "zhang@alibaba-inc.com", "company": "阿里巴巴"},
        {"name": "李华", "email": "lihua@gmail.com", "company": "某创业公司"},
        {"name": "王芳", "email": "wang@apple.com", "company": "Apple"},
    ]
    
    print("\n📥 AI 代理自动录入线索:")
    for lead_data in new_leads:
        result = json.loads(client.call_tool("create_lead", **lead_data))
        print(f"  ✅ 创建: {lead_data['name']} @ {lead_data['company']}")
        
        # 自动评分
        score_result = json.loads(client.call_tool("score_lead", lead_id=result["lead_id"]))
        print(f"     评分: {score_result['score']} 分 (等级: {score_result['grade']})")
    
    # 3. 获取所有线索
    all_leads = json.loads(client.call_tool("get_leads"))
    print(f"\n📊 当前线索库: {len(all_leads)} 条")
    for lead in all_leads:
        print(f"  #{lead['id']} {lead['name']} - {lead['company']} (评分: {lead['score']})")
    
    client.close()
    print("\n✅ AI 代理工作流完成！全程无人工干预。")

# migration_checker.py
# 检查你的 SaaS 产品是否准备好迎接 AI 代理
from dataclasses import dataclass
from typing import List
import requests

@dataclass
class APICheckResult:
    endpoint: str
    method: str
    has_docs: bool
    has_auth: bool
    returns_json: bool
    rate_limit: int
    
    @property
    def ai_ready_score(self) -> int:
        """AI 代理就绪度评分 (0-100)"""
        score = 0
        if self.has_docs: score += 30    # 文档是 AI 理解 API 的关键
        if self.has_auth: score += 20    # 安全的认证机制
        if self.returns_json: score += 25 # JSON 是 AI 最易处理的格式
        if self.rate_limit > 100: score += 25  # 高频率调用能力
        return score

class HeadlessReadinessChecker:
    """检查 SaaS 产品的 Headless AI 就绪度"""
    
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.headers = {"Authorization": f"Bearer {api_key}"}
        self.results: List[APICheckResult] = []
    
    def check_endpoint(self, endpoint: str, method: str = "GET") -> APICheckResult:
        """检查单个 API 端点"""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        
        try:
            response = requests.request(method, url, headers=self.headers, timeout=10)
            
            result = APICheckResult(
                endpoint=endpoint,
                method=method,
                has_docs=self._check_docs(endpoint),
                has_auth=response.status_code != 401,
                returns_json="application/json" in response.headers.get("Content-Type", ""),
                rate_limit=int(response.headers.get("X-RateLimit-Limit", 0))
            )
            self.results.append(result)
            return result
        except requests.RequestException as e:
            return APICheckResult(
                endpoint=endpoint, method=method,
                has_docs=False, has_auth=False,
                returns_json=False, rate_limit=0
            )
    
    def _check_docs(self, endpoint: str) -> bool:
        """检查是否有 API 文档（简化版）"""
        # 实际实现中应该检查 OpenAPI/Swagger 规范
        return True
    
    def generate_report(self) -> str:
        """生成就绪度报告"""
        if not self.results:
            return "❌ 没有检查任何端点"
        
        avg_score = sum(r.ai_ready_score for r in self.results) / len(self.results)
        
        report = f"""
╔══════════════════════════════════════════╗
║   Headless AI 就绪度报告                  ║
╠══════════════════════════════════════════╣
║  检查端点数: {len(self.results):>20} ║
║  平均就绪度: {avg_score:>16.1f}% ║
╠══════════════════════════════════════════╣
"""
        
        for r in self.results:
            status = "✅" if r.ai_ready_score >= 80 else "⚠️" if r.ai_ready_score >= 50 else "❌"
            report += f"║  {status} {r.method} {r.endpoint:<25} {r.ai_ready_score:>3}% ║\n"
        
        report += "╚══════════════════════════════════════════╝"
        return report

# 使用示例
if __name__ == "__main__":
    checker = HeadlessReadinessChecker(
        base_url="https://api.your-saas.com/v1",
        api_key="your-api-key"
    )
    
    # 检查核心端点
    endpoints = [
        "/users", "/leads", "/deals", "/contacts",
        "/products", "/orders", "/analytics"
    ]
    
    for ep in endpoints:
        checker.check_endpoint(ep)
    
    print(checker.generate_report())