# Directory Structure
```
├── .gitignore
├── .python-version
├── LICENSE
├── main.py
├── pyproject.toml
├── README.md
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
1 | 3.13
2 |
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
1 | # Python-generated files
2 | __pycache__/
3 | *.py[oc]
4 | build/
5 | dist/
6 | wheels/
7 | *.egg-info
8 |
9 | # Virtual environments
10 | .venv
11 |
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # OPC UA MCP Server
2 |
3 | An MCP server that connects to OPC UA-enabled industrial systems, allowing AI agents to monitor, analyze, and control operational data in real time.
4 |
5 | This project is ideal for developers and engineers looking to bridge AI-driven workflows with industrial automation systems.
6 |
7 | 
8 | 
9 | 
10 |
11 | ## Features
12 |
13 | - **Read OPC UA Nodes**: Retrieve real-time values from industrial devices.
14 | - **Write to OPC UA Nodes**: Control devices by writing values to specified nodes.
15 | - **Browse nodes**: Request to list allopcua nodes
16 | - **Read multiple OPC UA Nodes**: Retrieve multiple real-time values from devices.
17 | - **Write to multiple OPC UA Nodes**: Control devices by writing values to multiple nodes.
18 | - **Seamless Integration**: Works with MCP clients like Claude Desktop for natural language interaction.
19 |
20 |
21 | ### Tools
22 | The server exposes five tools:
23 | - **`read_opcua_node`**:
24 | - **Description**: Read the value of a specific OPC UA node.
25 | - **Parameters**:
26 | - `node_id` (str): OPC UA node ID (e.g., `ns=2;i=2`).
27 | - **Returns**: A string with the node ID and its value (e.g., "Node ns=2;i=2 value: 42").
28 |
29 | - **`write_opcua_node`**:
30 | - **Description**: Write a value to a specific OPC UA node.
31 | - **Parameters**:
32 | - `node_id` (str): OPC UA node ID (e.g., `ns=2;i=3`).
33 | - `value` (str): Value to write (converted based on node type).
34 | - **Returns**: A success or error message (e.g., "Successfully wrote 100 to node ns=2;i=3").
35 |
36 | - **`Browse nodes`**:
37 | - **Description**: Read the value of a specific OPC UA node.
38 |
39 | - **`Read multiple OPC UA Nodes`**:
40 | - **Description**: Read the value of a specific OPC UA node.
41 |
42 | - **`Write to multiple OPC UA Nodes`**:
43 | - **Description**: Read the value of a specific OPC UA node.
44 |
45 |
46 | ### Example Prompts
47 |
48 | - "What’s the value of node ns=2;i=2?" → Returns the current value.
49 | - "Set node ns=2;i=3 to 100." → Writes 100 to the node.
50 |
51 | ## Installation
52 |
53 | ### Prerequisites
54 | - Python 3.13 or higher
55 | - An OPC UA server (e.g., a simulator or real industrial device)
56 |
57 | ### Install Dependencies
58 | Clone the repository and install the required Python packages:
59 |
60 | ```bash
61 | git clone https://github.com/kukapay/opcua-mcp.git
62 | cd opcua-mcp
63 | pip install mcp[cli] opcua cryptography
64 | ```
65 |
66 | ### MCP Client Configuration
67 |
68 | ```json
69 | {
70 | "mcpServers": {
71 | "opcua-mcp": {
72 | "command": "python",
73 | "args": ["path/to/opcua_mcp/main.py"],
74 | "env": {
75 | "OPCUA_SERVER_URL": "your-opc-ua-server-url"
76 | }
77 | }
78 | }
79 | }
80 | ```
81 |
82 |
83 | ## License
84 | This project is licensed under the MIT License. See the [LICENSE](LICENSE) file for details.
85 |
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [project]
2 | name = "opcua-mcp"
3 | version = "0.1.0"
4 | description = "An MCP server that connects AI agents to OPC UA-enabled industrial systems."
5 | readme = "README.md"
6 | requires-python = ">=3.12.2"
7 | dependencies = [
8 | "cryptography>=45.0.2",
9 | "mcp[cli]>=1.5.0",
10 | "opcua>=0.98.13",
11 | ]
12 |
```
--------------------------------------------------------------------------------
/main.py:
--------------------------------------------------------------------------------
```python
1 | from mcp.server.fastmcp import FastMCP, Context
2 | from opcua import Client
3 | from contextlib import asynccontextmanager
4 | from typing import AsyncIterator
5 | import asyncio
6 | import os
7 | from typing import List, Dict, Any
8 | from opcua import ua #
9 |
10 | server_url = os.getenv("OPCUA_SERVER_URL", "opc.tcp://localhost:4840")
11 |
12 |
13 | # Manage the lifecycle of the OPC UA client connection
14 | @asynccontextmanager
15 | async def opcua_lifespan(server: FastMCP) -> AsyncIterator[dict]:
16 | """Handle OPC UA client connection lifecycle."""
17 | client = Client(server_url)
18 | try:
19 | # Connect to OPC UA server synchronously, wrapped in a thread for async compatibility
20 | await asyncio.to_thread(client.connect)
21 | print("Connected to OPC UA server")
22 | yield {"opcua_client": client}
23 | finally:
24 | # Disconnect from OPC UA server on shutdown
25 | await asyncio.to_thread(client.disconnect)
26 | print("Disconnected from OPC UA server")
27 |
28 |
29 | # Create an MCP server instance
30 | mcp = FastMCP("OPCUA-Control", lifespan=opcua_lifespan)
31 |
32 |
33 | # Tool: Read the value of an OPC UA node
34 | @mcp.tool()
35 | def read_opcua_node(node_id: str, ctx: Context) -> str:
36 | """
37 | Read the value of a specific OPC UA node.
38 |
39 | Parameters:
40 | node_id (str): The OPC UA node ID in the format 'ns=<namespace>;i=<identifier>'.
41 | Example: 'ns=2;i=2'.
42 |
43 | Returns:
44 | str: The value of the node as a string, prefixed with the node ID.
45 | """
46 | client = ctx.request_context.lifespan_context["opcua_client"]
47 | node = client.get_node(node_id)
48 | value = node.get_value() # Synchronous call to get node value
49 | return f"Node {node_id} value: {value}"
50 |
51 |
52 | # Tool: Write a value to an OPC UA node
53 | @mcp.tool()
54 | def write_opcua_node(node_id: str, value: str, ctx: Context) -> str:
55 | """
56 | Write a value to a specific OPC UA node.
57 |
58 | Parameters:
59 | node_id (str): The OPC UA node ID in the format 'ns=<namespace>;i=<identifier>'.
60 | Example: 'ns=2;i=3'.
61 | value (str): The value to write to the node. Will be converted based on node type.
62 |
63 | Returns:
64 | str: A message indicating success or failure of the write operation.
65 | """
66 | client = ctx.request_context.lifespan_context["opcua_client"]
67 | node = client.get_node(node_id)
68 | try:
69 | # Convert value based on the node's current type
70 | current_value = node.get_value()
71 | python_typed_value = value
72 | if isinstance(current_value, float):
73 | python_typed_value = float(value)
74 | elif isinstance(current_value, int):
75 | python_typed_value = int(value)
76 | # use variant type of node
77 | node.set_value(python_typed_value, node.get_data_type_as_variant_type())
78 | return f"Successfully wrote {value} to node {node_id}"
79 | except Exception as e:
80 | return f"Error writing to node {node_id}: {str(e)}"
81 |
82 |
83 | @mcp.tool()
84 | def browse_opcua_node_children(node_id: str, ctx: Context) -> str:
85 | """
86 | Browse the children of a specific OPC UA node.
87 |
88 | Parameters:
89 | node_id (str): The OPC UA node ID to browse (e.g., 'ns=0;i=85' for Objects folder).
90 |
91 | Returns:
92 | str: A string representation of a list of child nodes, including their NodeId and BrowseName.
93 | Returns an error message on failure.
94 | """
95 | client = ctx.request_context.lifespan_context["opcua_client"]
96 | try:
97 | node = client.get_node(node_id)
98 | children = node.get_children()
99 |
100 | children_info = []
101 | for child in children:
102 | try:
103 | browse_name = child.get_browse_name()
104 | children_info.append(
105 | {
106 | "node_id": child.nodeid.to_string(),
107 | "browse_name": f"{browse_name.NamespaceIndex}:{browse_name.Name}",
108 | }
109 | )
110 | except Exception as e:
111 | children_info.append(
112 | {
113 | "node_id": child.nodeid.to_string(),
114 | "browse_name": f"Error getting name: {e}",
115 | }
116 | )
117 |
118 | # import json
119 | # return json.dumps(children_info, indent=2)
120 | return f"Children of {node_id}: {children_info!r}"
121 |
122 | except Exception as e:
123 | return f"Error Browse children of node {node_id}: {str(e)}"
124 |
125 |
126 | @mcp.tool()
127 | def read_multiple_opcua_nodes(node_ids: List[str], ctx: Context) -> str:
128 | """
129 | Read the values of multiple OPC UA nodes in a single request.
130 |
131 | Parameters:
132 | node_ids (List[str]): A list of OPC UA node IDs to read (e.g., ['ns=2;i=2', 'ns=2;i=3']).
133 |
134 | Returns:
135 | str: A string representation of a dictionary mapping node IDs to their values, or an error message.
136 | """
137 | client = ctx.request_context.lifespan_context["opcua_client"]
138 | try:
139 | nodes_to_read = [client.get_node(nid) for nid in node_ids]
140 | values = []
141 | # Iterate over each node in nodes_to_read
142 | for node in nodes_to_read:
143 | try:
144 | # Get the value of the current node
145 | value = node.get_value()
146 | # Append the value to the values list
147 | values.append(value)
148 | except Exception as e:
149 | # In case of an error, append the error message
150 | values.append(f"Error reading node {node.nodeid.to_string()}: {str(e)}")
151 |
152 | # Map node IDs to their corresponding values
153 | results = {
154 | node.nodeid.to_string(): value for node, value in zip(nodes_to_read, values)
155 | }
156 |
157 | return f"Read multiple nodes values: {results!r}"
158 |
159 | except ua.UaError as e:
160 | status_name = e.code_as_name() if hasattr(e, "code_as_name") else "Unknown"
161 | status_code_hex = f"0x{e.code:08X}" if hasattr(e, "code") else "N/A"
162 | return f"Error reading multiple nodes {node_ids}: OPC UA Error - Status: {status_name} ({status_code_hex})"
163 | except Exception as e:
164 | return f"Error reading multiple nodes {node_ids}: {type(e).__name__} - {str(e)}"
165 |
166 |
167 | @mcp.tool()
168 | def write_multiple_opcua_nodes(
169 | nodes_to_write: List[Dict[str, Any]], ctx: Context
170 | ) -> str:
171 | """
172 | Write values to multiple OPC UA nodes in a single request.
173 |
174 | Parameters:
175 | nodes_to_write (List[Dict[str, Any]]): A list of dictionaries, where each dictionary
176 | contains 'node_id' (str) and 'value' (Any).
177 | The value will be wrapped in an OPC UA Variant.
178 | Example: [{'node_id': 'ns=2;i=2', 'value': 10.5},
179 | {'node_id': 'ns=2;i=3', 'value': 'active'}]
180 |
181 | Returns:
182 | str: A message indicating the success or failure of the write operation.
183 | Returns status codes for each write attempt.
184 | """
185 | client = ctx.request_context.lifespan_context["opcua_client"]
186 |
187 | node_ids_for_error_msg = [
188 | item.get("node_id", "unknown_node") for item in nodes_to_write
189 | ]
190 |
191 | try:
192 | nodes = [client.get_node(item["node_id"]) for item in nodes_to_write]
193 |
194 | # Iterate over nodes and values to set each value individually
195 | status_report = []
196 | for node, item in zip(nodes, nodes_to_write):
197 | try:
198 | # Create a Variant from the value
199 | value_as_variant = ua.Variant(item["value"])
200 | # Set the value of the node
201 | current_value = node.get_value()
202 | if isinstance(current_value, (int, float)):
203 | node.set_value(float(value_as_variant.Value))
204 | else:
205 | node.set_value(value_as_variant.Value)
206 |
207 | status_report.append(
208 | {
209 | "node_id": item["node_id"],
210 | "value_written": item["value"],
211 | "status": "Success",
212 | }
213 | )
214 | except Exception as e:
215 | return f"Error writing to node {node}: {str(e)}"
216 | # Return the status report
217 | return f"Write multiple nodes results: {status_report!r}"
218 |
219 | except ua.UaError as e:
220 | status_name = e.code_as_name() if hasattr(e, "code_as_name") else "Unknown"
221 | status_code_hex = f"0x{e.code:08X}" if hasattr(e, "code") else "N/A"
222 | return f"Error writing multiple nodes {node_ids_for_error_msg}: OPC UA Error - Status: {status_name} ({status_code_hex})"
223 | except Exception as e:
224 | return f"Error writing multiple nodes {node_ids_for_error_msg}: {type(e).__name__} - {str(e)}"
225 |
226 |
227 | # Run the server
228 | if __name__ == "__main__":
229 | mcp.run()
230 |
```