""" 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)