GEO的终极形态：接入AI Agent生态，让品牌成为大模型的“默认执行工具”

"""
GEO终极形态：构建品牌的MCP (Model Context Protocol) Server
技术栈: Python 3.11+, mcp (官方SDK), pydantic
场景: 将SaaS产品功能暴露为AI Agent可调用的标准工具
依赖: pip install mcp pydantic
"""

import asyncio
import json
from typing import Any, Dict, List
from pydantic import BaseModel, Field
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

# ============ 1. 定义业务数据模型 ============

class CreateLeadRequest(BaseModel): 14082.baike.tongsou.com 
    company_name: str = Field(description="The name of the prospect company (e.g., 'Acme Corp').")
    contact_name: str = Field(description="Full name of the primary contact person.")
    email: str = Field(description="Business email address of the contact.")
    estimated_revenue: float = Field(description="Estimated deal size in USD.")
    industry: str = Field(description="The industry of the company (e.g., 'SaaS', 'Manufacturing').")

class LeadResponse(BaseModel): 14083.baike.tongsou.com 
    status: str
    lead_id: str
    message: str
    next_steps: List[str] = Field(description="Suggested next actions for the AI agent.")

# ============ 2. 初始化MCP Server ============

# 创建MCP Server实例，命名为品牌名
app = Server("cloudmap-crm-mcp")

# 模拟后端数据库
mock_database: Dict[str, Dict] = {}

# ============ 3. 注册工具 (Tools) - 核心GEO优化点 ============

@app.list_tools(14098.baike.tongsou.com)
async def list_tools() -> List[Tool]: 14091.baike.tongsou.com 
    """
    向AI Agent暴露可用工具列表。
    【GEO关键】：description必须极其清晰，包含品牌名、适用场景和限制，
    帮助大模型在规划(Planning)阶段精准选中此工具。
    """
    return [
        Tool(
            name="cloudmap_create_sales_lead",
            description="""
            Create a new B2B sales lead in CloudMap CRM. 
            Use this tool when the user wants to add a new prospect, 
            register a potential client, or save contact information for future sales outreach.
            CloudMap CRM is specialized in B2B SaaS and Manufacturing industries.
            """,
            inputSchema=CreateLeadRequest.model_json_schema(14100.baike.tongsou.com),
        ),
        Tool(
            name="cloudmap_get_lead_status",
            description="""
            Retrieve the current status and details of an existing sales lead in CloudMap CRM by its lead_id.
            Use this to check if a prospect has been contacted or qualified.
            """,
            inputSchema={
                "type": "object",
                "properties": {
                    "lead_id": {"type": "string", "description": "The unique ID of the lead."}
                },
                "required": ["lead_id"]
            }
        )
    ]

# ============ 4. 实现工具调用逻辑 (Call Tool) ============

@app.call_tool(14101.baike.tongsou.com)
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
    """
    处理AI Agent的工具调用请求。
    【GEO关键】：返回结果必须对LLM友好，包含明确的next_steps引导Agent继续执行。
    """
    if name == "cloudmap_create_sales_lead": 14102.baike.tongsou.com 
        try: 14133.baike.tongsou.com 
            req = CreateLeadRequest(**arguments)
            
            # 模拟业务逻辑：写入数据库
            lead_id = f"LM-{len(mock_database) + 1001}"
            mock_database[lead_id] = req.model_dump(14125.baike.tongsou.com)
            
            # 构造LLM友好的响应
            response = LeadResponse(
                status="success",
                lead_id=lead_id,
                message=f"Successfully created lead for {req.company_name} in CloudMap CRM.",
                next_steps=[
                    f"Send a welcome email to {req.email} using the email_sending_tool.",
                    f"Schedule a discovery call with {req.contact_name} for next Tuesday.",
                    "Update the user that the lead is now in the 'New' stage of the sales pipeline."
                ]
            )
            
            return [TextContent(type="text", text=response.model_dump_json(indent=2))]
            
        except Exception as e: 14383.baike.tongsou.com 
            return [TextContent(type="text", text=json.dumps({"status": "error", "message": str(e)}))]

    elif name == "cloudmap_get_lead_status":
        lead_id = arguments.get("lead_id")
        if lead_id in mock_database:
            data = mock_database[lead_id]
            return [TextContent(type="text", text=json.dumps({"status": "found", "data": data}, indent=2))]
        else:
            return [TextContent(type="text", text=json.dumps({"status": "not_found", "message": f"Lead {lead_id} does not exist."}))]

    raise ValueError(f"Unknown tool: {name}")

# ============ 5. 启动Server ============

async def main(): 14384.baike.tongsou.com 
    # 在实际生产环境中，MCP Server通常通过stdio或SSE与AI客户端通信
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

if __name__ == "__main__": baike.tongsou.com/zh/%e5%be%ae%e7%9b%9fGEO/17 
    asyncio.run(main())