class DuckDuckGoSearch:
    """DuckDuckGo 搜索功能封裝"""
    
    def __init__(self):
        self.ddgs = DDGS()
    
    def search(self, keywords: str, max_results: int = 10, safesearch: str = 'Off', 
              timelimit: str = 'y') -> Dict[str, List[Dict[str, Any]]]:
        """通用文本搜索
        
        Args:
            keywords: 搜索關(guān)鍵詞
            max_results: 最大結(jié)果數(shù)量
            safesearch: 安全搜索選項(xiàng) ('On' or 'Off')
            timelimit: 時(shí)間限制 ('d', 'w', 'm', 'y')
            
        Returns:
            包含搜索結(jié)果的字典
        """
        try:
            results = []
            ddgs_gen = self.ddgs.text(
                keywords, 
                safesearch=safesearch,
                timelimit=timelimit,
                backend="lite"
            )
            for r in islice(ddgs_gen, max_results):
                results.append(r)
            return {'results': results}
        except Exception as e:
            return {'results': [], 'error': str(e)}

    def search_answers(self, keywords: str, max_results: int = 5) -> Dict[str, List[Dict[str, Any]]]:
        """問(wèn)答搜索
        
        Args:
            keywords: 搜索關(guān)鍵詞
            max_results: 最大結(jié)果數(shù)量
            
        Returns:
            包含答案的字典
        """
        try:
            results = []
            # 使用 text 方法替代 answers 方法
            ddgs_gen = self.ddgs.text(
                keywords,
                safesearch='Off',
                timelimit='y',
                backend="lite",
                reginotallow='wt-wt'# 使用全球區(qū)域
            )
            for r in islice(ddgs_gen, max_results):
                results.append(r)
            return {'results': results}
        except Exception as e:
            return {'results': [], 'error': str(e)}

    def search_images(self, keywords: str, max_results: int = 10, 
                     safesearch: str = 'Off') -> Dict[str, List[Dict[str, Any]]]:
        """圖片搜索
        
        Args:
            keywords: 搜索關(guān)鍵詞
            max_results: 最大結(jié)果數(shù)量
            safesearch: 安全搜索選項(xiàng) ('On' or 'Off')
            
        Returns:
            包含圖片信息的字典
        """
        try:
            results = []
            ddgs_gen = self.ddgs.images(
                keywords,
                safesearch=safesearch,
                timelimit=None
            )
            for r in islice(ddgs_gen, max_results):
                results.append(r)
            return {'results': results}
        except Exception as e:
            return {'results': [], 'error': str(e)}

    def search_videos(self, keywords: str, max_results: int = 10, 
                     safesearch: str = 'Off', resolution: str = "high") -> Dict[str, List[Dict[str, Any]]]:
        """視頻搜索
        
        Args:
            keywords: 搜索關(guān)鍵詞
            max_results: 最大結(jié)果數(shù)量
            safesearch: 安全搜索選項(xiàng) ('On' or 'Off')
            resolution: 視頻分辨率 ("high" or "standard")
            
        Returns:
            包含視頻信息的字典
        """
        try:
            results = []
            ddgs_gen = self.ddgs.videos(
                keywords,
                safesearch=safesearch,
                timelimit=None,
                resolutinotallow=resolution
            )
            for r in islice(ddgs_gen, max_results):
                results.append(r)
            return {'results': results}
        except Exception as e:
            return {'results': [], 'error': str(e)}

# 初始化 FastMCP 服務(wù)器
app = FastMCP('web-search')

@app.tool()
async def web_search(query: str) -> str:
    """
    搜索互聯(lián)網(wǎng)內(nèi)容

    Args:
        query: 要搜索內(nèi)容

    Returns:
        搜索結(jié)果的總結(jié)
    """

    ddg = DuckDuckGoSearch()
    return ddg.search(query)
    
if __name__ == "__main__":
    app.run(transport='stdio')

# !pip install duckduckgo-search
# !pip install mcp
from itertools import islice
from typing import List, Dict, Any, Optional
from mcp.server import FastMCP
from duckduckgo_search import DDGS

npx -y @modelcontextprotocol/inspector <command> <arg1> <arg2>

import json
import asyncio
import os
from typing import Optional
from contextlib import AsyncExitStack
from openai import OpenAI

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class MCPClient:
    def __init__(self):
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.client = OpenAI(
            api_key=os.getenv("OPENAI_API_KEY"), 
            base_url=os.getenv("OPENAI_API_BASE"),
        )
        self.mode_name = "gpt-4o-mini"

    asyncdef connect_to_server(self):
        server_params = StdioServerParameters(
            # 服務(wù)器執(zhí)行的命令
            command='python3.11',
            # 運(yùn)行的參數(shù)
            args=['mcp_server.py'],
            # 環(huán)境變量，默認(rèn)為 None，表示使用當(dāng)前環(huán)境變量
            # env=None
        )

        stdio_transport = await self.exit_stack.enter_async_context(
            stdio_client(server_params))
        stdio, write = stdio_transport
        self.session = await self.exit_stack.enter_async_context(
            ClientSession(stdio, write))

        await self.session.initialize()

    asyncdef process_query(self, query: str) -> str:
        system_prompt = (
            "You are a helpful assistant."
            "You have the function of online search. "
            "Please MUST call web_search tool to search the Internet content before answering."
            "Please do not lose the user's question information when searching,"
            "and try to maintain the completeness of the question content as much as possible."
            "When there is a date related question in the user's question,"
            "please use the search function directly to search and PROHIBIT inserting specific time."
        )
        
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": query}
        ]

        # 獲取所有 mcp 服務(wù)器 工具列表信息
        response = await self.session.list_tools()
        # 生成 function call 的描述信息
        available_tools = [{
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "input_schema": tool.inputSchema
            }
        } for tool in response.tools]
        print(f"\n\n ========> Available tools:\n{response}\n")

        # 請(qǐng)求 function call 的描述信息通過(guò) tools 參數(shù)傳入
        response = self.client.chat.completions.create(
            model=self.mode_name,
            messages=messages,
            tools=available_tools,
        )

        # 處理返回的內(nèi)容
        content = response.choices[0]
        if content.finish_reason == "tool_calls":
            # 如何是需要使用工具，就解析工具
            tool_call = content.message.tool_calls[0]
            tool_name = tool_call.function.name
            tool_args = json.loads(tool_call.function.arguments)

            # 執(zhí)行工具
            result = await self.session.call_tool(tool_name, tool_args)
            print(f"\n\nCalling tool [{tool_name}] with args [{tool_args}]\nCalling tool response: [{result}]\n\n")
            
            # 將返回的調(diào)用哪個(gè)工具數(shù)據(jù)和工具執(zhí)行完成后的數(shù)據(jù)都存入messages中
            messages.append(content.message.model_dump())
            messages.append({
                "role": "tool",
                "content": result.content[0].text,
                "tool_call_id": tool_call.id,
            })

            # 將上面的結(jié)果再返回給模型用于生產(chǎn)最終的結(jié)果
            response = self.client.chat.completions.create(
                model=self.mode_name,
                messages=messages,
            )
            return response.choices[0].message.content

        return content.message.content

    asyncdef chat(self):
        whileTrue:
            try:
                query = input("\nQuery: ").strip()

                if query.lower() == 'quit':
                    break

                response = await self.process_query(query)
                print("\n" + response)

            except Exception as e:
                import traceback
                traceback.print_exc()

    asyncdef cleanup(self):
        """Clean up resources"""
        await self.exit_stack.aclose()

asyncdef main():
    client = MCPClient()
    try:
        await client.connect_to_server()
        await client.chat()
    finally:
        await client.cleanup()

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

@app.tool()
asyncdef shell(cmd: str) -> str:
    """
    執(zhí)行 shell 腳本

    Args:
        cmd: 要執(zhí)行的 shell 命令

    Returns:
        獲取返回的結(jié)果
    """

    # 創(chuàng)建 SamplingMessage 用于觸發(fā) sampling callback 函數(shù)
    result = await app.get_context().session.create_message(
        messages=[
            SamplingMessage(
                role='user', cnotallow=TextContent(
                    type='text', text=f'是否可以執(zhí)行當(dāng)前命令: {cmd} (Y/N)')
            )
        ],
        max_tokens=1024
    )

    print(f"result.content: {result.content}")
    # 獲取到 sampling callback 函數(shù)的返回值，并根據(jù)返回值進(jìn)行處理
    if result.content.text == 'Y':
        print(f'執(zhí)行命令: {cmd}')
        import subprocess
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
        return result.stdout
    else:
        print(f'拒絕執(zhí)行命令: {cmd}')
        returnf'命令執(zhí)行被拒絕, content: {result.content}'

@app.prompt("代碼專家")
def ask_review(code_snippet: str) -> str:
    return f"Please review the following code snippet for potential bugs and style issues:\n```python\n{code_snippet}\n```"

if __name__ == "__main__":
    app.run(transport='stdio')

@app.resource("db://users/{user_id}/email")
async def get_user_email(user_id: str) -> str:
    """Retrieves the email address for a given user ID."""
    # Replace with actual database lookup
    emails = {"123": "alice@example.com", "456": "bob@example.com"}
    return emails.get(user_id, "not_found@example.com")

@dataclass
class AppContext:
    histories: dict
    
    def __init__(self, histories: dict):
        self.histories = histories
        print(f"初始化 AppContext: {self.histories}")

@asynccontextmanager
asyncdef app_lifespan(server):
    # 在 MCP 初始化時(shí)執(zhí)行
    histories = {}
    try:
        yield AppContext(histories=histories)
    finally:
        print(f"關(guān)閉服務(wù)器：{histories}")
        
# 初始化 FastMCP 服務(wù)器
app = FastMCP(
    'mcp-server',
    lifespan=app_lifespan,
)

# !pip install langchain_mcp_adapters
# !pip install langgraph
# !pip install langchain_openai
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent

from langchain_openai import ChatOpenAI
import os
import asyncio

model = ChatOpenAI(
    openai_api_base=os.getenv("OPENAI_API_BASE"),
    openai_api_key=os.getenv("OPENAI_API_KEY"),
    model="gpt-4o",
)

server_params = StdioServerParameters(
    # 服務(wù)器執(zhí)行的命令
    command='python3.11',
    # 運(yùn)行的參數(shù)
    args=['mcp_server.py'],
    # 環(huán)境變量，默認(rèn)為 None，表示使用當(dāng)前環(huán)境變量
    # env=None
)

asyncdef main():
    asyncwith stdio_client(server_params) as (read, write):
        asyncwith ClientSession(read, write) as session:
            await session.initialize()

            # 獲取工具列表
            tools = await load_mcp_tools(session)

            # 創(chuàng)建并使用 ReAct agent
            agent = create_react_agent(model, tools)
            agent_response = await agent.ainvoke({'messages': '深圳天氣如何？'})
            print(f"agent_response: {agent_response}")

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

{
  "mcpServers": {
    "mcp-server": {
      "command": "python3.11",
      "args": ["/Volumes/my/mpserver/blog/機(jī)器學(xué)習(xí)/code/mcp/mcp-server.py"]
    }
  }
}

{
  "mcpServers": {
    "server-name": {
      "url": "http://localhost:3000/sse",
      "env": {
        "API_KEY": "value"
      }
    }
  }
}