mcp-server-pacman

Instructions

Implementation Reference

• src/mcp_server_pacman/providers/terraform.py:132-199 (handler)

  Core handler function that performs HTTP request to Terraform Registry API to fetch versions, extracts the latest version (first in list), and formats the response dictionary.

  @async_cached(http_cache)
  async def get_latest_terraform_module_version(name: str) -> Dict:
      """Get the latest version of a Terraform module.
  
      The name parameter should be in the format: namespace/name/provider
      """
      # Split the module name into namespace, name, and provider
      parts = name.split("/")
      if len(parts) != 3:
          raise McpError(
              ErrorData(
                  code=INTERNAL_ERROR,
                  message="Invalid Terraform module name format. Expected: namespace/name/provider",
              )
          )
  
      namespace, module_name, provider = parts
  
      async with httpx.AsyncClient() as client:
          response = await client.get(
              f"https://registry.terraform.io/v1/modules/{namespace}/{module_name}/{provider}/versions",
              headers={"Accept": "application/json", "User-Agent": DEFAULT_USER_AGENT},
              follow_redirects=True,
          )
  
          if response.status_code != 200:
              raise McpError(
                  ErrorData(
                      code=INTERNAL_ERROR,
                      message=f"Failed to get module versions from Terraform Registry - status code {response.status_code}",
                  )
              )
  
          try:
              data = response.json()
              modules = data.get("modules", [])
  
              if not modules:
                  raise McpError(
                      ErrorData(
                          code=INTERNAL_ERROR,
                          message=f"No versions found for module {name}",
                      )
                  )
  
              # The latest version is typically the first one in the list
              latest = modules[0]
  
              result = {
                  "name": f"{namespace}/{module_name}/{provider}",
                  "namespace": namespace,
                  "provider": provider,
                  "module": module_name,
                  "version": latest.get("version", ""),
                  "published_at": latest.get("published_at", ""),
                  "source": latest.get("source", ""),
              }
              return result
          except McpError:
              raise
          except Exception as e:
              raise McpError(
                  ErrorData(
                      code=INTERNAL_ERROR,
                      message=f"Failed to parse Terraform Registry module versions: {str(e)}",
                  )
              )

• src/mcp_server_pacman/models/package_models.py:73-79 (schema)

  Pydantic BaseModel defining the input schema for the tool: requires 'name' field with description of expected format.

  class TerraformModuleLatestVersion(BaseModel):
      """Parameters for getting the latest version of a Terraform module."""
  
      name: Annotated[
          str, Field(description="Module name (format: namespace/name/provider)")
      ]

• src/mcp_server_pacman/server.py:90-94 (registration)

  Registers the MCP tool in list_tools() with name, description, and references the input schema.

  Tool(
      name="terraform_module_latest_version",
      description="Get the latest version of a Terraform module",
      inputSchema=TerraformModuleLatestVersion.model_json_schema(),
  ),

• src/mcp_server_pacman/server.py:365-393 (handler)

  Dispatch handler in @server.call_tool(): validates arguments using schema, calls the provider function, formats response as TextContent, handles errors.

  elif name == "terraform_module_latest_version":
      try:
          args = TerraformModuleLatestVersion(**arguments)
          logger.debug(f"Validated terraform module latest version args: {args}")
      except ValueError as e:
          logger.error(
              f"Invalid terraform module latest version parameters: {str(e)}"
          )
          raise McpError(ErrorData(code=INVALID_PARAMS, message=str(e)))
  
      logger.info(f"Getting latest version for Terraform module {args.name}")
  
      try:
          info = await get_latest_terraform_module_version(args.name)
          logger.info(
              f"Successfully retrieved latest version for Terraform module {args.name}"
          )
          return [
              TextContent(
                  type="text",
                  text=f"Latest version for Terraform module {args.name}:\n{json.dumps(info, indent=2)}",
              )
          ]
      except McpError as e:
          logger.error(
              f"Error getting latest version for Terraform module: {str(e)}"
          )
          raise