鸿蒙操作系统-MCP服务器

<div align="center"> <h1>HarmonyOS MCP 服务器</h1>

<a href='LICENSE'><img src='https://img.shields.io/badge/License-MIT-orange'></a>     <a><img src='https://img.shields.io/badge/python-3.13-blue'></a>

</div> <div align="center"> <img style="max-width: 500px; width: 60%;" width="1111" alt="image" src="https://gips0.baidu.com/it/u=1065694064,454919044&fm=3081&app=3081&f=PNG?w=2222&h=1326" /> </div>

简介

这是一个用于操作HarmonyOS设备的MCP服务器。

https://github.com/user-attachments/assets/7af7f5af-e8c6-4845-8d92-cd0ab30bfe17

快速开始

安装

1. 克隆此仓库

git clone https://github.com/XixianLiang/HarmonyOS-mcp-server.git
cd HarmonyOS-mcp-server

2. 设置环境。

uv python install 3.13
uv sync

使用方法

1. Claude Desktop

您可以使用 Claude Desktop 来尝试我们的工具。

2. Openai SDK

您也可以使用 openai-agents SDK 来尝试MCP服务器。这里有一个示例

"""
示例：使用Openai-agents SDK调用HarmonyOS-mcp-server
"""
import asyncio
import os

from agents import Agent, Runner, gen_trace_id, trace
from agents.mcp import MCPServerStdio, MCPServer

async def run(mcp_server: MCPServer):
    agent = Agent(
        name="Assistant",
        instructions="使用工具来操作HarmonyOS设备并完成任务。",
        mcp_servers=[mcp_server],
    )

    message = "在手机上启动应用`设置`"
    print(f"运行中: {message}")
    result = await Runner.run(starting_agent=agent, input=message)
    print(result.final_output)


async def main():

    # 使用异步上下文管理器初始化服务器
    async with MCPServerStdio(
        params={
            "command": "<...>/bin/uv",
            "args": [
                "--directory",
                "<...>/harmonyos-mcp-server",
                "run",
                "server.py"
            ]
        }
    ) as server:
        trace_id = gen_trace_id()
        with trace(workflow_name="MCP HarmonyOS", trace_id=trace_id):
            print(f"查看跟踪: https://platform.openai.com/traces/trace?trace_id={trace_id}\n")
            await run(server)

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

3. Langchain

您可以使用 LangGraph，一个灵活的LLM代理框架来设计您的工作流。这里有一个示例

"""
langgraph_mcp.py
"""

server_params = StdioServerParameters(
    command="/home/chad/.local/bin/uv",
    args=["--directory",
          ".",
          "run",
          "server.py"],
    
)


# 此函数会使用langgraph构建自己的代理工作流
async def create_graph(session):
    llm = ChatOllama(model="qwen2.5:7b", temperature=0)
    # !!!load_mcp_tools 是一个langchain包函数，它将mcp集成到langchain中。
    # !!!bind_tools 函数使您的llm能够访问您的mcp工具
    tools = await load_mcp_tools(session)
    llm_with_tool = llm.bind_tools(tools)

    
    system_prompt = await load_mcp_prompt(session, "system_prompt")
    prompt_template = ChatPromptTemplate.from_messages([
        ("system", system_prompt[0].content),
        MessagesPlaceholder("messages")
    ])
    chat_llm = prompt_template | llm_with_tool

    # 状态管理
    class State(TypedDict):
        messages: Annotated[List[AnyMessage], add_messages]

    # 节点
    def chat_node(state: State) -> State:
        state["messages"] = chat_llm.invoke({"messages": state["messages"]})
        return state

    # 构建图
    # 图像是您的代理的工作流程。
    # 如果您想了解更多关于langgraph的基本知识，请参考此链接 (https://langchain-ai.github.io/langgraph/tutorials/get-started/1-build-basic-chatbot/#3-add-a-node)
    graph_builder = StateGraph(State)
    graph_builder.add_node("chat_node", chat_node)
    graph_builder.add_node("tool_node", ToolNode(tools=tools))
    graph_builder.add_edge(START, "chat_node")
    graph_builder.add_conditional_edges("chat_node", tools_condition, {"tools": "tool_node", "__end__": END})
    graph_builder.add_edge("tool_node", "chat_node")
    graph = graph_builder.compile(checkpointer=MemorySaver())
    return graph



async def main():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            config = RunnableConfig(thread_id=1111, recursion_limit=15)
            # 在图中使用MCP服务器
            agent = await create_graph(session)

            while True:
                message = input("用户: ")
                try:
                    response = await agent.ainvoke({"messages": message}, config=config)
                    print("AI: "+response["messages"][-1].content)
                except RecursionError:
                    result = None
                    logging.error("图递归限制已达到。")


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

在 server.py 中编写系统提示

"""
server.py
"""
@mcp.prompt()
def system_prompt() -> str:
    """系统提示描述"""
    return """
    您是一个AI助手，如有需要可以使用工具。
    """

使用 load_mcp_prompt 函数从mcp服务器获取您的提示。

"""
langgraph_mcp.py
"""
prompts = await load_mcp_prompt(session, "system_prompt")