[Tools] Add two new tools - list_runs and run_function (#3)

This PR adds two new tools:
- `run_function`: allows users to trigger a function through the MCP
- `list_runs`: provides a detailed summary of previous runs

Author: Eyal-Danieli

src/mcp_server/mlrun_client/api_client.py

@@ -33,6 +33,7 @@
 
 import mlrun.db.auth_utils
 import mlrun.db.httpdb
+import mlrun_pipelines.models
 
 if typing.TYPE_CHECKING:
     import mlrun.artifacts
@@ -118,6 +119,23 @@ async def get_artifact(self, project_name: str, *args, **kwargs) -> mlrun.artifa
         project = self.__get_project(project_name)
         return await asyncio.to_thread(project.get_artifact, *args, **kwargs)
 
+
+    async def run_function(self, project_name: str, *args, **kwargs) -> typing.Union[mlrun.model.RunObject, mlrun_pipelines.models.PipelineNodeWrapper]:
+        """Runs a function in the given project
+
+        Returns the function runtime after execution.
+        """
+        project = self.__get_project(project_name)
+        return await asyncio.to_thread(project.run_function, *args, **kwargs)
+
+    async def list_runs(self, project_name: str, *args, **kwargs) -> list:
+        """Lists runs of a given project
+
+        Returns a list of runs.
+        """
+        project = self.__get_project(project_name)
+        return await asyncio.to_thread(project.list_runs, *args, **kwargs)
+
     async def list_model_endpoints(self, project_name: str, *args, **kwargs) -> list[dict]:
         """Lists model endpoints of a given project
 

src/mcp_server/server.py

@@ -21,9 +21,9 @@
 
 from __future__ import annotations
 
+import datetime
 from typing import Annotated
 
-import datetime
 import fastmcp
 from fastmcp.server.dependencies import AccessToken, get_access_token
 from pydantic import Field
@@ -101,6 +101,288 @@ async def get_artifact_uri(
     async with get_mlrun_api_client() as client:
         return (await client.get_artifact(project_name, key=key, tag=tag)).uri
 
+@mcp.tool
+async def run_function(
+    function_name: Annotated[str, Field(description="Name of the function to run (e.g., 'my_function')")],
+    project_name: project_name_field,
+    handler: Annotated[
+        str | None,
+        Field(
+            description="Name of the function handler",
+            default=None,
+        ),
+    ] = None,
+    name: Annotated[
+        str,
+        Field(
+            description="Execution name",
+            default="",
+        ),
+    ] = "",
+    params: Annotated[
+        dict | None,
+        Field(
+            description="Input parameters (dict) to pass to the function (e.g., {'param1': 'value1'})",
+            default=None,
+        ),
+    ] = None,
+    inputs: Annotated[
+        dict | None,
+        Field(
+            description="Input objects to pass to the handler (e.g., {'dataset': 'store://datasets/my_dataset'})",
+            default=None,
+        ),
+    ] = None,
+    labels: Annotated[
+        dict | None,
+        Field(
+            description="Labels to tag the job/run with (e.g., {'key1': 'val1', 'key2': 'val2'})",
+            default=None,
+        ),
+    ] = None,
+    workdir: Annotated[
+        str,
+        Field(
+            description="Working directory of the executed job and the default path for artifact inputs",
+            default="",
+        ),
+    ] = "",
+    schedule: Annotated[
+        str | None,
+        Field(
+            description="Standard crontab expression string for scheduling (e.g., '0 0 * * *' for daily at midnight)",
+            default=None,
+        ),
+    ] = None,
+    output_path: Annotated[
+        str | None,
+        Field(
+            description="Path to store artifacts, when running in a workflow this will be set automatically",
+            default=None,
+        ),
+    ] = None,
+    local: Annotated[
+        bool | None,
+        Field(
+            description="Run the function locally vs on the runtime/cluster",
+            default=None,
+        ),
+    ] = None,
+    watch: Annotated[
+        bool,
+        Field(
+            description="Watch/follow run log",
+            default=True,
+        ),
+    ] = True,
+) -> dict:
+    """
+    Run a function in the specified project.
+
+    This tool executes an MLRun function and returns information about the run.
+    The function can be run locally or on a remote runtime/cluster.
+    """
+    async with get_mlrun_api_client() as client:
+        run_result = await client.run_function(
+            project_name,
+            function=function_name,
+            handler=handler,
+            name=name,
+            params=params,
+            inputs=inputs,
+            labels=labels,
+            workdir=workdir,
+            schedule=schedule,
+            output_path=output_path,
+            local=local,
+            watch=watch,
+        )
+
+        # Extract relevant information from the run result
+        return {
+            "uid": run_result.metadata.uid if hasattr(run_result, 'metadata') else None,
+            "name": run_result.metadata.name if hasattr(run_result, 'metadata') else None,
+            "project": run_result.metadata.project if hasattr(run_result, 'metadata') else None,
+            "state": run_result.status.state if hasattr(run_result, 'status') else None,
+            "results": run_result.status.results if hasattr(run_result, 'status') else None,
+            "artifacts": [
+                {
+                    "key": artifact.get("key"),
+                    "kind": artifact.get("kind"),
+                }
+                for artifact in (run_result.status.artifacts if hasattr(run_result, 'status') and run_result.status.artifacts else [])
+            ],
+        }
+
+
+@mcp.tool
+async def list_runs(
+    project_name: project_name_field,
+    name: Annotated[
+        str | None,
+        Field(
+            description="Name of the run to retrieve",
+            default=None,
+        ),
+    ] = None,
+    uid: Annotated[
+        str | None,
+        Field(
+            description="Unique ID of the run (single UID as string)",
+            default=None,
+        ),
+    ] = None,
+    labels: Annotated[
+        str | None,
+        Field(
+            description="Filter runs by labels. Format: 'key1=value1,key2=value2' or 'key1,key2' for key existence",
+            default=None,
+        ),
+    ] = None,
+    states: Annotated[
+        str | None,
+        Field(
+            description="Comma-separated list of states to filter by (e.g., 'completed,running')",
+            default=None,
+        ),
+    ] = None,
+    sort: Annotated[
+        bool,
+        Field(
+            description="Whether to sort the result according to their start time",
+            default=True,
+        ),
+    ] = True,
+    iter: Annotated[
+        bool,
+        Field(
+            description="If True return runs from all iterations. Otherwise, return only runs whose iter is 0",
+            default=False,
+        ),
+    ] = False,
+    start_time_from: Annotated[
+        str | None,
+        Field(
+            description="Filter by run start time from (ISO format with timezone, e.g., '2025-11-24T00:00:00Z')",
+            default=None,
+        ),
+    ] = None,
+    start_time_to: Annotated[
+        str | None,
+        Field(
+            description="Filter by run start time to (ISO format with timezone, e.g., '2025-11-27T23:59:59Z')",
+            default=None,
+        ),
+    ] = None,
+    last_update_time_from: Annotated[
+        str | None,
+        Field(
+            description="Filter by run last update time from (ISO format with timezone)",
+            default=None,
+        ),
+    ] = None,
+    last_update_time_to: Annotated[
+        str | None,
+        Field(
+            description="Filter by run last update time to (ISO format with timezone)",
+            default=None,
+        ),
+    ] = None,
+    end_time_from: Annotated[
+        str | None,
+        Field(
+            description="Filter by run end time from (ISO format with timezone)",
+            default=None,
+        ),
+    ] = None,
+    end_time_to: Annotated[
+        str | None,
+        Field(
+            description="Filter by run end time to (ISO format with timezone)",
+            default=None,
+        ),
+    ] = None,
+) -> list[dict]:
+    """
+    Retrieve a list of runs.
+
+    The default returns the runs from the last week, partitioned by name.
+    To override the default, specify any filter.
+    """
+    # Parse labels parameter
+    labels_param = None
+    if labels:
+        # Check if it's key=value format or just keys
+        if '=' in labels:
+            # Parse as dict: "key1=value1,key2=value2"
+            labels_param = {}
+            for label in labels.split(','):
+                if '=' in label:
+                    key, value = label.split('=', 1)
+                    labels_param[key.strip()] = value.strip()
+                else:
+                    # Key existence check
+                    labels_param[label.strip()] = None
+        else:
+            # Parse as list of keys: "key1,key2"
+            labels_param = [key.strip() for key in labels.split(',')]
+
+    # Parse states parameter
+    states_list = None
+    if states:
+        states_list = [state.strip() for state in states.split(',')]
+
+    # Parse datetime parameters
+    def parse_datetime(dt_str: str | None) -> datetime.datetime | None:
+        if not dt_str:
+            return None
+        try:
+            return datetime.datetime.fromisoformat(dt_str.replace('Z', '+00:00'))
+        except ValueError:
+            raise ValueError(
+                f"Invalid datetime format: '{dt_str}'. "
+                "Expected ISO format with timezone (e.g., '2025-11-24T00:00:00Z')"
+            )
+
+    start_time_from_dt = parse_datetime(start_time_from)
+    start_time_to_dt = parse_datetime(start_time_to)
+    last_update_time_from_dt = parse_datetime(last_update_time_from)
+    last_update_time_to_dt = parse_datetime(last_update_time_to)
+    end_time_from_dt = parse_datetime(end_time_from)
+    end_time_to_dt = parse_datetime(end_time_to)
+
+    async with get_mlrun_api_client() as client:
+        runs = await client.list_runs(
+            project_name,
+            name=name,
+            uid=uid,
+            labels=labels_param,
+            states=states_list,
+            sort=sort,
+            iter=iter,
+            start_time_from=start_time_from_dt,
+            start_time_to=start_time_to_dt,
+            last_update_time_from=last_update_time_from_dt,
+            last_update_time_to=last_update_time_to_dt,
+            end_time_from=end_time_from_dt,
+            end_time_to=end_time_to_dt,
+        )
+
+    # Convert RunList to list of dicts
+    return [
+        {
+            "uid": run.get("metadata", {}).get("uid"),
+            "name": run.get("metadata", {}).get("name"),
+            "project": run.get("metadata", {}).get("project"),
+            "state": run.get("status", {}).get("state"),
+            "start_time": run.get("status", {}).get("start_time"),
+            "last_update": run.get("status", {}).get("last_update"),
+            "labels": run.get("metadata", {}).get("labels", {}),
+            "results": run.get("status", {}).get("results"),
+        }
+        for run in runs
+    ]
+
 
 @mcp.tool
 async def get_model_endpoints(