get_code

Instructions

Implementation Reference

• app/tools/vibe_tools.py:395-419 (handler)

  The _execute_get_code method is the internal handler that validates required parameters (query_id, version_id), calls vibe_client.get_code(), and formats the JSON response.

  async def _execute_get_code(
      self,
      tool_input: dict[str, Any],
      log: structlog.stdlib.BoundLogger,
  ) -> list[TextContent]:
      """Execute get_code tool."""
      required = ["query_id", "version_id"]
      missing = [p for p in required if p not in tool_input]
      if missing:
          raise MCPToolValidationError(
              f"Missing required parameters: {', '.join(missing)}"
          )
  
      response = await self.vibe_client.get_code(
          query_id=tool_input["query_id"],
          version_id=tool_input["version_id"],
      )
  
      result = {
          "status": "success",
          "data": response.model_dump(),
      }
  
      log.info("Tool executed successfully", result=result)
      return [TextContent(type="text", text=json.dumps(result, indent=2))]

• app/services/vibe_api.py:364-416 (handler)

  The get_code method in VibeAPIClient creates a GetCodeRequest, makes the HTTP GET call to the GET_CODE endpoint, and returns a validated GetCodeResponse.

  async def get_code(
      self,
      query_id: str,
      version_id: str,
  ) -> GetCodeResponse:
      """
      Get code by query/code ID.
  
      Args:
          query_id: Query/Code ID
          version_id: Version ID
  
      Returns:
          GetCodeResponse with code details
  
      Raises:
          MCPToolExecutionError: If API call fails
      """
      log = self.logger.bind(
          method="get_code",
          query_id=query_id,
          version_id=version_id,
      )
  
      try:
          # Create request object (validates input)
          request = GetCodeRequest(
              query_id=query_id,
              version_id=version_id,
          )
  
          log.debug("Fetching code", request=request.model_dump(by_alias=True))
  
          # Make API call
          response_data = await self.http_client.get(
              endpoint=API_ENDPOINTS["GET_CODE"],
              params=request.model_dump(
                  by_alias=True,
                  exclude_none=True,
              ),
          )
  
          # Validate and transform response
          response = GetCodeResponse(**response_data)
          log.debug("Code retrieved successfully")
  
          return response
  
      except Exception as e:
          log.error("Failed to get code", error=str(e))
          raise MCPToolExecutionError(
              f"Failed to retrieve code: {str(e)}"
          ) from e

• app/models/code.py:13-26 (schema)

  GetCodeRequest model with query_id (QueryID) and version_id (versionId) fields.

  class GetCodeRequest(BaseModel):
      """Request model for GetCodeById endpoint."""
  
      model_config = ConfigDict(extra="forbid", populate_by_name=True)
  
      query_id: str = Field(
          alias="QueryID",
          description="Query/Code ID to retrieve",
      )
      version_id: str = Field(
          alias="versionId",
          description="Version ID",
      )

• app/models/code.py:63-76 (schema)

  GetCodeResponse model containing success flag, optional CodeDto, and optional message.

  class GetCodeResponse(BaseModel):
      """Response model for GetCodeById endpoint."""
  
      model_config = ConfigDict(extra="allow")
  
      success: bool = Field(description="Success indicator")
      code: Optional[CodeDto] = Field(
          default=None,
          description="Code details",
      )
      message: Optional[str] = Field(
          default=None,
          description="Response message",
      )

• app/tools/registration.py:156-170 (registration)

  Registration of the 'get_code' MCP tool using @mcp.tool decorator with name='get_code' and description, wiring it to _execute_get_code.

  @mcp.tool(
      name="get_code",
      description="Get code information from the Vibe system",
  )
  async def get_code(query_id: str, version_id: str) -> str:
      try:
          result = await _tools()._execute_get_code(
              {"query_id": query_id, "version_id": version_id},
              logger.bind(tool="get_code"),
          )
          return result[0].text
      except Exception as e:
          log = logger.bind(tool="get_code")
          log.error("Tool execution failed", error=str(e))
          raise MCPToolExecutionError(f"Tool execution failed: {str(e)}") from e