SideFX Houdini customization package
git clone https://git.lucas.co/hou-control.git
houdini-agent/mcpserver/houdini_mcp.py (4K)
1 #!/usr/bin/env python3
2 import os
3 import rpyc
4 from mcp.server.fastmcp import FastMCP
5
6 HOUDINI_HOST = os.environ.get("HOUDINI_MCP_HOST", "localhost")
7 HOUDINI_PORT = int(os.environ.get("HOUDINI_MCP_PORT", "18811"))
8
9 mcp = FastMCP("houdini")
10
11
12 _REMOTE_RUNNER_SRC = r'''
13 def _hcrun(code):
14 import io, contextlib, traceback, sys
15 main_ns = sys.modules["__main__"].__dict__
16 try:
17 import hou
18 if "hou" not in main_ns:
19 main_ns["hou"] = hou
20 except ImportError:
21 pass
22
23 def _work():
24 buf = io.StringIO()
25 result_repr = None
26 error = None
27 try:
28 with contextlib.redirect_stdout(buf), contextlib.redirect_stderr(buf):
29 try:
30 result = eval(compile(code, "<mcp>", "eval"), main_ns)
31 if result is not None:
32 try:
33 result_repr = repr(result)
34 except Exception as _e:
35 result_repr = "<unrepr-able: " + repr(_e) + ">"
36 except SyntaxError:
37 exec(compile(code, "<mcp>", "exec"), main_ns)
38 except Exception:
39 error = traceback.format_exc()
40 return buf.getvalue(), result_repr, error
41
42 # rpyc's ThreadedServer serves this call on a connection thread, and HOM
43 # is not thread-safe: a hou call off Houdini's main thread segfaults it
44 # inside SWIG. Hand the work to the main-thread pump uiready installs,
45 # published on hou.session so this injected source needs no import of
46 # its own. No pump (hython, no UI, startup not finished) means no UI
47 # thread to protect, so run inline.
48 _hc_call = getattr(
49 getattr(sys.modules.get("hou"), "session", None),
50 "_hc_main_thread_call",
51 None,
52 )
53 if _hc_call is None:
54 return _work()
55 return _hc_call(_work)
56 '''
57
58
59 @mcp.tool()
60 def houdini_eval(code: str) -> str:
61 """Run Python code inside the running Houdini session.
62
63 The `hou` module and anything in __main__ is available. Captures stdout/stderr.
64 If the code is a single expression, its repr is returned too.
65 Requires Houdini to be running with hrpyc.start_server() called (happens
66 automatically in uiready.py for this project).
67 """
68 try:
69 conn = rpyc.classic.connect(HOUDINI_HOST, HOUDINI_PORT)
70 except ConnectionRefusedError:
71 return (
72 f"ERROR: Could not connect to Houdini at {HOUDINI_HOST}:{HOUDINI_PORT}. "
73 "Is Houdini running with hrpyc started?"
74 )
75
76 try:
77 conn.execute(_REMOTE_RUNNER_SRC)
78 stdout, result_repr, error = conn.namespace["_hcrun"](code)
79 stdout = str(stdout) if stdout else ""
80 result_repr = str(result_repr) if result_repr else None
81 error = str(error) if error else None
82 finally:
83 try:
84 conn.close()
85 except Exception:
86 pass
87
88 parts = []
89 if stdout:
90 parts.append(stdout.rstrip())
91 if result_repr is not None:
92 parts.append(f"=> {result_repr}")
93 if error:
94 parts.append(error.rstrip())
95 return "\n".join(parts) if parts else "(no output)"
96
97
98 @mcp.tool()
99 def houdini_ping() -> str:
100 """Check whether the Houdini RPC server is reachable."""
101 try:
102 conn = rpyc.classic.connect(HOUDINI_HOST, HOUDINI_PORT)
103 try:
104 # conn.modules.hou.* would evaluate on the connection thread;
105 # go through the runner so it lands on Houdini's main thread.
106 conn.execute(_REMOTE_RUNNER_SRC)
107 _out, result_repr, error = conn.namespace["_hcrun"](
108 "(hou.applicationVersionString(), hou.hipFile.path())"
109 )
110 if error:
111 return f"Error: {str(error).rstrip()}"
112 return f"OK — {str(result_repr)}"
113 finally:
114 conn.close()
115 except ConnectionRefusedError:
116 return f"Unreachable at {HOUDINI_HOST}:{HOUDINI_PORT}"
117 except Exception as e:
118 return f"Error: {e!r}"
119
120
121 if __name__ == "__main__":
122 mcp.run()