kabudadada commited on
Commit
69f64a8
·
1 Parent(s): 23f0e72

feat(esm mcp): add get_prediction_pdb tool to download saved PDB (supports latest)

Browse files
esm/mcp_output/mcp_plugin/mcp_service.py CHANGED
@@ -150,20 +150,8 @@ def predict_structure(sequence: str):
150
  pdb_io = io.StringIO(response.text)
151
  structure = parser.get_structure("esmfold_prediction", pdb_io)
152
 
153
- # Prefer persistent storage in Space (e.g., /data), fallback to local package dir
154
- default_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "predictions")
155
- preferred_dir = "/data/predictions"
156
- predictions_dir = preferred_dir
157
- try:
158
- os.makedirs(predictions_dir, exist_ok=True)
159
- # Test write permission with a temp touch
160
- _probe_path = os.path.join(predictions_dir, ".write_probe")
161
- with open(_probe_path, "w") as _probe:
162
- _probe.write("ok")
163
- os.remove(_probe_path)
164
- except Exception:
165
- predictions_dir = default_dir
166
- os.makedirs(predictions_dir, exist_ok=True)
167
 
168
  timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
169
  pdb_filename = f"prediction_{timestamp}.pdb"
@@ -265,6 +253,52 @@ def validate_protein_sequence(sequence: str):
265
  }
266
 
267
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
268
  def create_app():
269
  """
270
  Create and return a FastMCP instance.
 
150
  pdb_io = io.StringIO(response.text)
151
  structure = parser.get_structure("esmfold_prediction", pdb_io)
152
 
153
+ predictions_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "predictions")
154
+ os.makedirs(predictions_dir, exist_ok=True)
 
 
 
 
 
 
 
 
 
 
 
 
155
 
156
  timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
157
  pdb_filename = f"prediction_{timestamp}.pdb"
 
253
  }
254
 
255
 
256
+ @mcp.tool(name="get_prediction_pdb", description="Download saved PDB by filename or latest")
257
+ def get_prediction_pdb(filename: str):
258
+ """Return PDB content saved under predictions directory.
259
+
260
+ Parameters:
261
+ filename (str): Filename in predictions directory. Use "latest" to fetch the most recent file.
262
+
263
+ Returns:
264
+ dict: success/result/error. result contains { filename, pdb_content, path }
265
+ """
266
+ try:
267
+ import glob
268
+ predictions_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "predictions")
269
+ if not os.path.isdir(predictions_dir):
270
+ return {"success": False, "result": None, "error": f"Predictions dir not found: {predictions_dir}"}
271
+
272
+ target_path: str
273
+ if filename.strip().lower() == "latest":
274
+ files = sorted(
275
+ glob.glob(os.path.join(predictions_dir, "*.pdb")),
276
+ key=lambda p: os.path.getmtime(p),
277
+ reverse=True,
278
+ )
279
+ if not files:
280
+ return {"success": False, "result": None, "error": "No PDB files found"}
281
+ target_path = files[0]
282
+ filename = os.path.basename(target_path)
283
+ else:
284
+ # prevent path traversal
285
+ safe_name = os.path.basename(filename)
286
+ target_path = os.path.join(predictions_dir, safe_name)
287
+ if not os.path.isfile(target_path):
288
+ return {"success": False, "result": None, "error": f"File not found: {safe_name}"}
289
+
290
+ with open(target_path, "r") as f:
291
+ content = f.read()
292
+
293
+ return {
294
+ "success": True,
295
+ "result": {"filename": filename, "pdb_content": content, "path": target_path},
296
+ "error": None,
297
+ }
298
+ except Exception as e:
299
+ return {"success": False, "result": None, "error": str(e)}
300
+
301
+
302
  def create_app():
303
  """
304
  Create and return a FastMCP instance.