HarmonyOS MCP Server

1<div align="center">
2<h1>HarmonyOS MCP Server</h1>
3 
4 <a href='LICENSE'><img src='https://img.shields.io/badge/License-MIT-orange'></a> &nbsp;&nbsp;&nbsp;
5 <a><img src='https://img.shields.io/badge/python-3.13-blue'></a>
6</div>
7 
8<div align="center">
9    <img style="max-width: 500px; width: 60%;" width="1111" alt="image" src="https://github.com/user-attachments/assets/7c2e6879-f583-48d7-b467-c4c6d99c5fab" />
10</div>
11 
12## Intro
13 
14This is a MCP server for manipulating harmonyOS Device.
15 
16 
17https://github.com/user-attachments/assets/7af7f5af-e8c6-4845-8d92-cd0ab30bfe17
18 
19 
20## Quick Start
21 
22### Installation
23 
241. Clone this repo
25   
26```bash
27git clone https://github.com/XixianLiang/HarmonyOS-mcp-server.git
28cd HarmonyOS-mcp-server
29```
30 
312. Setup the envirnment.
32 
33```bash
34uv python install 3.13
35uv sync
36```
37 
38### Usage
39 
40 
41#### 1.Claude Desktop
42 
43You can use [Claude Desktop](https://modelcontextprotocol.io/quickstart/user) to try our tool.
44 
45#### 2.Openai SDK
46You can also use [openai-agents SDK](https://openai.github.io/openai-agents-python/mcp/) to try the mcp server. Here's an example
47 
48```python
49"""
50Example: Use Openai-agents SDK to call HarmonyOS-mcp-server
51"""
52import asyncio
53import os
54 
55from agents import Agent, Runner, gen_trace_id, trace
56from agents.mcp import MCPServerStdio, MCPServer
57 
58async def run(mcp_server: MCPServer):
59    agent = Agent(
60        name="Assistant",
61        instructions="Use the tools to manipulate the HarmonyOS device and finish the task.",
62        mcp_servers=[mcp_server],
63    )
64 
65    message = "Launch the app `settings` on the phone"
66    print(f"Running: {message}")
67    result = await Runner.run(starting_agent=agent, input=message)
68    print(result.final_output)
69 
70 
71async def main():
72 
73    # Use async context manager to initialize the server
74    async with MCPServerStdio(
75        params={
76            "command": "<...>/bin/uv",
77            "args": [
78                "--directory",
79                "<...>/harmonyos-mcp-server",
80                "run",
81                "server.py"
82            ]
83        }
84    ) as server:
85        trace_id = gen_trace_id()
86        with trace(workflow_name="MCP HarmonyOS", trace_id=trace_id):
87            print(f"View trace: https://platform.openai.com/traces/trace?trace_id={trace_id}\n")
88            await run(server)
89 
90if __name__ == "__main__":
91    asyncio.run(main())
92```
93 
94#### 3.Langchain
95You can use [LangGraph](https://langchain-ai.github.io/langgraph/concepts/why-langgraph/), a flexible LLM agent framework to design your workflows. Here's an example
96 
97```python
98"""
99langgraph_mcp.py
100"""
101 
102server_params = StdioServerParameters(
103    command="/home/chad/.local/bin/uv",
104    args=["--directory",
105          ".",
106          "run",
107          "server.py"],
108    
109)
110 
111 
112#This fucntion would use langgraph to build your own agent workflow
113async def create_graph(session):
114    llm = ChatOllama(model="qwen2.5:7b", temperature=0)
115    #!!!load_mcp_tools is a langchain package function that integrates the mcp into langchain.
116    #!!!bind_tools fuction enable your llm to access your mcp tools
117    tools = await load_mcp_tools(session)
118    llm_with_tool = llm.bind_tools(tools)
119 
120    
121    system_prompt = await load_mcp_prompt(session, "system_prompt")
122    prompt_template = ChatPromptTemplate.from_messages([
123        ("system", system_prompt[0].content),
124        MessagesPlaceholder("messages")
125    ])
126    chat_llm = prompt_template | llm_with_tool
127 
128    # State Management
129    class State(TypedDict):
130        messages: Annotated[List[AnyMessage], add_messages]
131 
132    # Nodes
133    def chat_node(state: State) -> State:
134        state["messages"] = chat_llm.invoke({"messages": state["messages"]})
135        return state
136 
137    # Building the graph
138    # graph is like a workflow of your agent.
139    #If you want to know more langgraph basic,reference this link (https://langchain-ai.github.io/langgraph/tutorials/get-started/1-build-basic-chatbot/#3-add-a-node)
140    graph_builder = StateGraph(State)
141    graph_builder.add_node("chat_node", chat_node)
142    graph_builder.add_node("tool_node", ToolNode(tools=tools))
143    graph_builder.add_edge(START, "chat_node")
144    graph_builder.add_conditional_edges("chat_node", tools_condition, {"tools": "tool_node", "__end__": END})
145    graph_builder.add_edge("tool_node", "chat_node")
146    graph = graph_builder.compile(checkpointer=MemorySaver())
147    return graph
148 
149 
150 
151 
152 
153async def main():
154    async with stdio_client(server_params) as (read, write):
155        async with ClientSession(read, write) as session:
156            await session.initialize()
157 
158            config = RunnableConfig(thread_id=1234,recursion_limit=15)
159            # Use the MCP Server in the graph
160            agent = await create_graph(session)
161 
162            while True:
163                message = input("User: ")
164                try:
165                    response = await agent.ainvoke({"messages": message}, config=config)
166                    print("AI: "+response["messages"][-1].content)
167                except RecursionError:
168                    result = None
169                    logging.error("Graph recursion limit reached.")
170 
171 
172if __name__ == "__main__":
173    asyncio.run(main())
174```
175 
176Write the system prompt in `server.py`
177 
178```python
179"""
180server.py
181"""
182@mcp.prompt()
183def system_prompt() -> str:
184    """System prompt description"""
185    return """
186    You are an AI assistant use the tools if needed.
187    """
188```
189Use `load_mcp_prompt` function to get your prompt from mcp server.
190```python
191"""
192langgraph_mcp.py
193"""
194prompts = await load_mcp_prompt(session, "system_prompt")
195```
196