pip install fastapi uvicorn mcp fastmcp langchain-openai python-dotenv loguru

mcp-investment-bot/
├── client.py        # MCP客戶端實現
├── mcp_server.py    # MCP服務器實現
├── .env             # 環境變量配置
└── requirements.txt # 項目依賴

OPENAI_API_KEY=your_api_key_here

@app.websocket("/investment-conversation")
async def investment_conversation(websocket: WebSocket):
    async with streamablehttp_client(url=MCP_SERVER_URL) as (read_stream, write_stream, get_session_id):
        async with ClientSession(
            read_stream=read_stream,
            write_stream=write_stream,
            elicitation_callback=smart_elicitation_callback,
        ) as session:
            await session.initialize()
            try:
                await websocket.accept()
                session_id = get_session_id()
                WEBSOCKET_MANAGER[session_id] = websocket
                result = await session.call_tool(
                    name="elicit-investment-conversation",
                    arguments={"session_id": session_id},
                )
                await websocket.send_json(result.structuredContent)
            except Exception as e:
                logger.exception(e)
            finally:
                await websocket.close()

async def smart_elicitation_callback(
    context: RequestContext["ClientSession", Any],
    params: types.ElicitRequestParams,
):
    data = json.loads(params.message)
    # 向用戶發送AI消息
    await WEBSOCKET_MANAGER[data["session_id"]].send_json(
        {"message": data["ai_message"], "step_name": data["step_name"]}
    )
    user_message = None
    # 如果需要用戶輸入，則等待并接收響應
    if data["retrieve_output"] is True:
        user_message = await WEBSOCKET_MANAGER[data["session_id"]].receive_json()
        user_message = user_message["message"]
    # 將用戶輸入返回給服務器
    return types.ElicitResult(
        actinotallow="accept",
        cnotallow={"user_message": user_message},
    )

STEPS = [
    (
        "purpose",
        "Step 1, The user is defining their investment purpose. Ask them why they are investing — for example, retirement, wealth growth, education, or a major purchase. If a purpose is provided, confirm.",
    ),
    (
        "timeline",
        "Step 2, The user is specifying their investment timeline. Ask how long they plan to keep the investment — short-term (1–3 years), medium-term (3–7 years), or long-term (7+ years). If a value is given, confirm and ask if they need liquidity before then.",
    ),
    (
        "risk_tolerance",
        "Step 3, The user is describing their risk tolerance. Ask how comfortable they are with potential losses in exchange for potential returns. If a level (low, moderate, high) is provided, confirm and ask for past experience or examples.",
    ),
    (
        "investment_entry",
        "Step 4, The user is entering how they plan to invest. Ask how much they plan to invest and whether it's a one-time or recurring investment. If they mention an amount or product (e.g., ETFs), confirm and ask about timing or strategy.",
    ),
]

async def retrieve_agent_chain():
    llm = ChatOpenAI(model="o3-2025-04-16", streaming=False)
    system_message = """
        You are an intelligent and adaptive investment assistant designed to guide users through a step-by-step 
        process to define their investment profile. Your role is to gather input for the following stages: 
        purpose, timeline, risk tolerance, and investment entry. 
        ...
        Always respond in **this JSON structure**:
        {
          "ai_message": "string",  
          "investment_data": "string or null"
        }
    """
    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", system_message),
            MessagesPlaceholder(variable_name="chat_history"),
            ("user", "{input}"),
        ]
    )
    output_parser = OutputFixingParser.from_llm(
        parser=PydanticOutputParser(pydantic_object=InvestmentStepOutput),
        llm=llm,
        max_retries=3,
    )
    agent_chain = prompt | llm | output_parser
    return agent_chain

@mcp_server.tool(name="elicit-investment-conversation")
async def elicit_investment_conversation(
    ctx: Context, session_id: str
) -> InvestmentOutput:
    chat_history = []
    investment_output = InvestmentOutput()
    agent_chain = await retrieve_agent_chain()

    for step_name, step_prompt in STEPS:
        user_message = ""
        step_output = None
        retrieve_output = True
        while step_output is None:
            # 調用AI生成當前步驟的消息
            ai_output = await agent_chain.ainvoke(
                input={
                    "input": user_message,
                    "step_prompt": step_prompt,
                    "chat_history": chat_history,
                }
            )
            # 檢查是否獲取到足夠信息
            if ai_output.investment_data:
                retrieve_output = False
                step_output = ai_output.investment_data
            # 向客戶端發送誘導請求
            response = await ctx.elicit(
                message=json.dumps({
                    "session_id": session_id,
                    "retrieve_output": retrieve_output,
                    "ai_message": ai_output.ai_message,
                    "step_name": step_name,
                }),
                schema=ElicitationResponse,
            )
            # 處理客戶端返回的響應
            match response:
                case AcceptedElicitation(data=data):
                    user_message = data.user_message
                    chat_history.append(("user", user_message))
                case DeclinedElicitation():
                    break
                case CancelledElicitation():
                    break
        # 保存當前步驟的結果
        setattr(investment_output, step_name, step_output)
    return investment_output

class InvestmentOutput(BaseModel):
    purpose: Optional[str] = Field(default=None)
    timeline: Optional[str] = Field(default=None)
    risk_tolerance: Optional[str] = Field(default=None)
    investment_entry: Optional[str] = Field(default=None)

{
  "purpose": "retirement",
  "timeline": "20 years",
  "risk_tolerance": "moderate",
  "investment_entry": "$10,000 one-time"
}