280 lines
8.5 KiB
Python
280 lines
8.5 KiB
Python
"""
|
|
Shepard MCP server — tools for the Shepard research data platform.
|
|
Uses the v1 REST API. Auth: X-API-KEY header.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from fastmcp import FastMCP
|
|
|
|
BASE_URL = os.environ.get("SHEPARD_BASE_URL", "https://shepard-api.nuclide.systems/shepard/api")
|
|
API_KEY = os.environ.get("SHEPARD_API_KEY", "")
|
|
|
|
mcp = FastMCP(
|
|
"shepard",
|
|
instructions=(
|
|
"Tools for the Shepard research data management platform. "
|
|
"Use search_shepard for broad discovery, list_collections to navigate the research hierarchy, "
|
|
"and publish_entity to mint a PID and finalize a data object or collection. "
|
|
"Collections contain DataObjects; DataObjects own Containers (file, timeseries, structured-data) "
|
|
"and lab journal entries."
|
|
),
|
|
)
|
|
|
|
|
|
def _client() -> httpx.AsyncClient:
|
|
headers = {"Accept": "application/json", "Content-Type": "application/json"}
|
|
if API_KEY:
|
|
headers["X-API-KEY"] = API_KEY
|
|
return httpx.AsyncClient(base_url=BASE_URL, headers=headers, timeout=30.0)
|
|
|
|
|
|
async def _get(path: str, params: dict | None = None) -> Any:
|
|
async with _client() as c:
|
|
r = await c.get(path, params={k: v for k, v in (params or {}).items() if v is not None})
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
async def _post(path: str, body: dict) -> Any:
|
|
async with _client() as c:
|
|
r = await c.post(path, json=body)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
async def _patch(path: str, body: dict) -> Any:
|
|
async with _client() as c:
|
|
r = await c.patch(path, json=body)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Collections
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@mcp.tool()
|
|
async def list_collections(page: int = 0, size: int = 20) -> list:
|
|
"""List all research collections the caller may read."""
|
|
return await _get("/collections", {"page": page, "size": size})
|
|
|
|
|
|
@mcp.tool()
|
|
async def get_collection(collection_id: int) -> dict:
|
|
"""Get a single collection by its numeric ID."""
|
|
return await _get(f"/collections/{collection_id}")
|
|
|
|
|
|
@mcp.tool()
|
|
async def create_collection(name: str, description: str | None = None) -> dict:
|
|
"""Create a new research collection."""
|
|
body: dict = {"name": name}
|
|
if description:
|
|
body["description"] = description
|
|
return await _post("/collections", body)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data Objects
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@mcp.tool()
|
|
async def list_data_objects(
|
|
collection_id: int,
|
|
name: str | None = None,
|
|
page: int = 0,
|
|
size: int = 20,
|
|
) -> list:
|
|
"""List DataObjects inside a collection.
|
|
|
|
Args:
|
|
collection_id: Numeric ID of the collection.
|
|
name: Optional name filter (substring match).
|
|
page: Page offset.
|
|
size: Items per page.
|
|
"""
|
|
return await _get(
|
|
f"/collections/{collection_id}/dataObjects",
|
|
{"name": name, "page": page, "size": size},
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
async def get_data_object(collection_id: int, data_object_id: int) -> dict:
|
|
"""Get a specific DataObject by its numeric IDs."""
|
|
return await _get(f"/collections/{collection_id}/dataObjects/{data_object_id}")
|
|
|
|
|
|
@mcp.tool()
|
|
async def patch_data_object(
|
|
collection_id: int,
|
|
data_object_id: int,
|
|
fields: dict,
|
|
) -> dict:
|
|
"""Partially update a DataObject (RFC 7396 JSON Merge Patch).
|
|
|
|
Args:
|
|
collection_id: Numeric collection ID.
|
|
data_object_id: Numeric DataObject ID.
|
|
fields: Fields to update, e.g. {"name": "New name", "description": "..."}.
|
|
"""
|
|
return await _patch(
|
|
f"/collections/{collection_id}/dataObjects/{data_object_id}",
|
|
fields,
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
async def create_data_object(
|
|
collection_id: int,
|
|
name: str,
|
|
description: str | None = None,
|
|
) -> dict:
|
|
"""Create a new DataObject inside a collection."""
|
|
body: dict = {"name": name}
|
|
if description:
|
|
body["description"] = description
|
|
return await _post(f"/collections/{collection_id}/dataObjects", body)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Containers (File, Timeseries, Structured Data)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@mcp.tool()
|
|
async def list_file_containers(page: int = 0, size: int = 20) -> list:
|
|
"""List all file containers the caller may read."""
|
|
return await _get("/fileContainers", {"page": page, "size": size})
|
|
|
|
|
|
@mcp.tool()
|
|
async def list_timeseries_containers(page: int = 0, size: int = 20) -> list:
|
|
"""List all timeseries containers the caller may read."""
|
|
return await _get("/timeseriesContainers", {"page": page, "size": size})
|
|
|
|
|
|
@mcp.tool()
|
|
async def list_structured_data_containers(page: int = 0, size: int = 20) -> list:
|
|
"""List all structured-data containers the caller may read."""
|
|
return await _get("/structuredDataContainers", {"page": page, "size": size})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Publish / Finalize
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@mcp.tool()
|
|
async def publish_entity(kind: str, app_id: str) -> dict:
|
|
"""Publish (finalize) an entity — mints a persistent identifier (PID) via the active Minter.
|
|
|
|
Use this when a data object or collection is ready for public reference / citation.
|
|
|
|
Args:
|
|
kind: Entity kind: 'dataObjects', 'collections', 'fileContainers',
|
|
'timeseriesContainers', 'structuredDataContainers'.
|
|
app_id: appId (UUID) of the entity to publish.
|
|
"""
|
|
async with _client() as c:
|
|
r = await c.post(f"/v2/{kind}/{app_id}/publish")
|
|
r.raise_for_status()
|
|
return r.json() if r.content else {"status": "published"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Lab Journal
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@mcp.tool()
|
|
async def list_lab_journal(data_object_id: int) -> list:
|
|
"""List lab journal entries attached to a DataObject.
|
|
|
|
Args:
|
|
data_object_id: Numeric ID of the DataObject.
|
|
"""
|
|
return await _get("/labJournalEntries", {"dataObjectId": data_object_id})
|
|
|
|
|
|
@mcp.tool()
|
|
async def get_lab_journal_entry(entry_id: int) -> dict:
|
|
"""Get a single lab journal entry by its numeric ID."""
|
|
return await _get(f"/labJournalEntries/{entry_id}")
|
|
|
|
|
|
@mcp.tool()
|
|
async def create_lab_journal_entry(
|
|
collection_id: int,
|
|
data_object_id: int,
|
|
title: str,
|
|
content: str,
|
|
) -> dict:
|
|
"""Create a new lab journal entry attached to a DataObject.
|
|
|
|
Args:
|
|
collection_id: Numeric ID of the parent collection.
|
|
data_object_id: Numeric ID of the DataObject to attach the entry to.
|
|
title: Entry title.
|
|
content: Entry content (Markdown supported).
|
|
"""
|
|
return await _post("/labJournalEntries", {
|
|
"collectionId": collection_id,
|
|
"dataObjectId": data_object_id,
|
|
"title": title,
|
|
"content": content,
|
|
})
|
|
|
|
|
|
@mcp.tool()
|
|
async def update_lab_journal_entry(
|
|
entry_id: int,
|
|
title: str | None = None,
|
|
content: str | None = None,
|
|
) -> dict:
|
|
"""Partially update a lab journal entry."""
|
|
body: dict = {}
|
|
if title is not None:
|
|
body["title"] = title
|
|
if content is not None:
|
|
body["content"] = content
|
|
return await _patch(f"/labJournalEntries/{entry_id}", body)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Search
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@mcp.tool()
|
|
async def search_data_objects(
|
|
query: str,
|
|
collection_id: int | None = None,
|
|
) -> list:
|
|
"""Search DataObjects by name (full-text).
|
|
|
|
Args:
|
|
query: Search string.
|
|
collection_id: Optional numeric collection ID to scope the search.
|
|
"""
|
|
scope: dict = {"traversalRules": []}
|
|
if collection_id is not None:
|
|
scope["collectionId"] = collection_id
|
|
return await _post("/search", {
|
|
"searchParams": {"query": query, "queryType": "DataObject"},
|
|
"scopes": [scope],
|
|
})
|
|
|
|
|
|
@mcp.tool()
|
|
async def search_collections(query: str) -> list:
|
|
"""Search Collections by name."""
|
|
return await _post("/search", {
|
|
"searchParams": {"query": query, "queryType": "Collection"},
|
|
"scopes": [{"traversalRules": []}],
|
|
})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)
|