get_deployments

Instructions

Implementation Reference

• src/mcp_prefect/deployment.py:16-90 (handler)

  The @mcp.tool decorated handler function that implements the get_deployments tool logic. It queries Prefect deployments using filters and returns results with UI links.

  @mcp.tool
  async def get_deployments(
      limit: Optional[int] = None,
      offset: Optional[int] = None,
      flow_name: Optional[str] = None,
      name: Optional[str] = None,
      tags: Optional[List[str]] = None,
      is_schedule_active: Optional[bool] = None,
      work_queue_name: Optional[str] = None,
  ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
      """
      Get a list of deployments with optional filtering.
      
      Args:
          limit: Maximum number of deployments to return
          offset: Number of deployments to skip
          flow_name: Filter by flow name
          name: Filter by deployment name
          tags: Filter by tags
          is_schedule_active: Filter by schedule active status
          work_queue_name: Filter by work queue name
          
      Returns:
          A list of deployments with their details
      """
      try:
          async with get_client() as client:
              # Build deployment filter
              deployment_filter = None
              if any([name, tags, is_schedule_active, work_queue_name]):
                  from prefect.client.schemas.filters import DeploymentFilter
                  
                  filter_dict = {}
                  if name:
                      filter_dict["name"] = {"like_": f"%{name}%"}
                  if tags:
                      filter_dict["tags"] = {"all_": tags}
                  if is_schedule_active is not None:
                      filter_dict["is_schedule_active"] = {"eq_": is_schedule_active}
                  if work_queue_name:
                      filter_dict["work_queue_name"] = {"eq_": work_queue_name}
                  
                  deployment_filter = DeploymentFilter(**filter_dict)
              
              # Build flow filter if flow_name is specified
              flow_filter = None
              if flow_name:
                  from prefect.client.schemas.filters import FlowFilter
                  
                  flow_filter = FlowFilter(name={"like_": f"%{flow_name}%"})
              
              # Query using proper filter objects
              deployments = await client.read_deployments(
                  deployment_filter=deployment_filter,
                  flow_filter=flow_filter,
                  limit=limit,
                  offset=offset,
              )
              
              # Add UI links to each deployment
              deployments_result = {
                  "deployments": [
                      {
                          **deployment.model_dump(),
                          "ui_url": get_deployment_url(str(deployment.id))
                      }
                      for deployment in deployments
                  ]
              }
              
              return [types.TextContent(type="text", text=str(deployments_result))]
      except Exception as e:
          # Add proper error handling
          return [types.TextContent(type="text", text=str({"error": str(e)}))]

• src/mcp_prefect/main.py:46-48 (registration)

  Conditional import of the deployment module in main.py, which triggers registration of all deployment tools including get_deployments via their @mcp.tool decorators.

  if APIType.DEPLOYMENT.value in apis:
      info("Loading Deployment API...")
      from . import deployment

• src/mcp_prefect/deployment.py:11-13 (helper)

  Helper function used within get_deployments (and other deployment tools) to generate UI URLs for deployments.

  def get_deployment_url(deployment_id: str) -> str:
      base_url = PREFECT_API_URL.replace("/api", "")
      return f"{base_url}/deployments/{deployment_id}"